Repository: ambari Updated Branches: refs/heads/trunk d3429491a -> 21e37b164
AMBARI-22505 : Kafka service check fails when using a non-root user in kerberized environment (ydavis via mradhakrishnan) Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/21e37b16 Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/21e37b16 Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/21e37b16 Branch: refs/heads/trunk Commit: 21e37b164f87edeb21b7533af1558e4758d5ee60 Parents: d342949 Author: Madhuvanthi Radhakrishnan <mradhakrish...@hortonworks.com> Authored: Thu Nov 30 11:31:51 2017 -0800 Committer: Madhuvanthi Radhakrishnan <mradhakrish...@hortonworks.com> Committed: Thu Nov 30 11:31:51 2017 -0800 ---------------------------------------------------------------------- .../KAFKA/0.8.1/package/scripts/service_check.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/21e37b16/ambari-server/src/main/resources/common-services/KAFKA/0.8.1/package/scripts/service_check.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1/package/scripts/service_check.py b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1/package/scripts/service_check.py index cb66f98..38a7f24 100644 --- a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1/package/scripts/service_check.py +++ b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1/package/scripts/service_check.py @@ -19,8 +19,10 @@ limitations under the License. """ from resource_management.libraries.script.script import Script from resource_management.libraries.functions.validate import call_and_match_output +from resource_management.core import shell from resource_management.libraries.functions.format import format from resource_management.core.logger import Logger +from resource_management.core.exceptions import Fail from resource_management.core import sudo import subprocess @@ -31,15 +33,20 @@ class ServiceCheck(Script): # TODO, Kafka Service check should be more robust , It should get all the broker_hosts # Produce some messages and check if consumer reads same no.of messages. - + kafka_config = self.read_kafka_config() topic = "ambari_kafka_service_check" create_topic_cmd_created_output = "Created topic \"ambari_kafka_service_check\"." create_topic_cmd_exists_output = "Topic \"ambari_kafka_service_check\" already exists." source_cmd = format("source {conf_dir}/kafka-env.sh") - topic_exists_cmd = format("{kafka_home}/bin/kafka-topics.sh --zookeeper {kafka_config[zookeeper.connect]} --topic {topic} --list") - topic_exists_cmd_p = subprocess.Popen(topic_exists_cmd.split(" "), stdout=subprocess.PIPE, stderr=subprocess.PIPE) - topic_exists_cmd_out, topic_exists_cmd_err = topic_exists_cmd_p.communicate() + topic_exists_cmd = format(source_cmd + " ; " + "{kafka_home}/bin/kafka-topics.sh --zookeeper {kafka_config[zookeeper.connect]} --topic {topic} --list") + topic_exists_cmd_code, topic_exists_cmd_out = shell.call(topic_exists_cmd, logoutput=True, quiet=False, user=params.kafka_user) + + if topic_exists_cmd_code > 0: + raise Fail("Error encountered when attempting to list topics: {0}".format(topic_exists_cmd_out)) + + + # run create topic command only if the topic doesn't exists delete_topic_cmd = format("{kafka_home}/bin/kafka-topics.sh --zookeeper {kafka_config[zookeeper.connect]} --delete --topic {topic}") create_topic_cmd = format("{kafka_home}/bin/kafka-topics.sh --zookeeper {kafka_config[zookeeper.connect]} --create --topic {topic} --partitions 1 --replication-factor 1")