AMBARI-20434. Enhance Kafka Service Check. Additional changes (WangJie via dlysnichenko)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/4b3cbef3 Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/4b3cbef3 Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/4b3cbef3 Branch: refs/heads/branch-feature-AMBARI-12556 Commit: 4b3cbef39edb37b8b393dd5b19c968223a282281 Parents: 242d9e7 Author: Lisnichenko Dmitro <[email protected]> Authored: Thu May 11 13:27:06 2017 +0300 Committer: Lisnichenko Dmitro <[email protected]> Committed: Thu May 11 13:27:06 2017 +0300 ---------------------------------------------------------------------- .../KAFKA/0.10.0.3.0/configuration/kafka-broker.xml | 2 +- .../0.10.0.3.0/package/scripts/service_check.py | 15 ++++++++++----- 2 files changed, 11 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/4b3cbef3/ambari-server/src/main/resources/common-services/KAFKA/0.10.0.3.0/configuration/kafka-broker.xml ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/common-services/KAFKA/0.10.0.3.0/configuration/kafka-broker.xml b/ambari-server/src/main/resources/common-services/KAFKA/0.10.0.3.0/configuration/kafka-broker.xml index b62b986..46c14c0 100644 --- a/ambari-server/src/main/resources/common-services/KAFKA/0.10.0.3.0/configuration/kafka-broker.xml +++ b/ambari-server/src/main/resources/common-services/KAFKA/0.10.0.3.0/configuration/kafka-broker.xml @@ -527,7 +527,7 @@ </property> <property> <name>delete.topic.enable</name> - <value>false</value> + <value>true</value> <description>Enables delete topic. Delete topic through the admin tool will have no effect if this config is turned off</description> <on-ambari-upgrade add="false"/> </property> http://git-wip-us.apache.org/repos/asf/ambari/blob/4b3cbef3/ambari-server/src/main/resources/common-services/KAFKA/0.10.0.3.0/package/scripts/service_check.py ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/resources/common-services/KAFKA/0.10.0.3.0/package/scripts/service_check.py b/ambari-server/src/main/resources/common-services/KAFKA/0.10.0.3.0/package/scripts/service_check.py index 0f3a417..cb66f98 100644 --- a/ambari-server/src/main/resources/common-services/KAFKA/0.10.0.3.0/package/scripts/service_check.py +++ b/ambari-server/src/main/resources/common-services/KAFKA/0.10.0.3.0/package/scripts/service_check.py @@ -40,12 +40,17 @@ class ServiceCheck(Script): 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() - # run create topic command only if the topic doesn't exists - if topic not in topic_exists_cmd_out: - create_topic_cmd = format("{kafka_home}/bin/kafka-topics.sh --zookeeper {kafka_config[zookeeper.connect]} --create --topic {topic} --partitions 1 --replication-factor 1") + + 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") + if topic in topic_exists_cmd_out: + # run delete topic and recreate the topic command only if the topic exists + command = source_cmd + " ; " + delete_topic_cmd + ";" + create_topic_cmd + else: + # run create topic command command = source_cmd + " ; " + create_topic_cmd - Logger.info("Running kafka create topic command: %s" % command) - call_and_match_output(command, format("({create_topic_cmd_created_output})|({create_topic_cmd_exists_output})"), "Failed to check that topic exists", user=params.kafka_user) + Logger.info("Running kafka create topic command: %s" % command) + call_and_match_output(command, format("({create_topic_cmd_created_output})|({create_topic_cmd_exists_output})"), "Failed to check that topic exists", user=params.kafka_user) def read_kafka_config(self): import params
