|
@@ -18,20 +18,20 @@ class Topic():
|
|
|
self.config_entries = config_entries
|
|
|
|
|
|
|
|
|
-client = pykafka.KafkaClient(hosts="192.168.3.17:9092")
|
|
|
-print(client.topics)
|
|
|
-for i in client.brokers.values():
|
|
|
- try:
|
|
|
- i.create_topics(topic_reqs=(Topic("tag_sink_topic"),))
|
|
|
- except Exception as e:
|
|
|
- if re.search("41", str(e)):
|
|
|
- print("该broker 不是 leader,交由下一个broker创建")
|
|
|
- elif re.search("7", str(e)):
|
|
|
- print("创建完成")
|
|
|
- break
|
|
|
- elif re.search("36", str(e)):
|
|
|
- print("topic 已存在")
|
|
|
- break
|
|
|
- else:
|
|
|
- raise e
|
|
|
+client = pykafka.KafkaClient(hosts="192.168.36.199:9092")
|
|
|
print(client.topics)
|
|
|
+# for i in client.brokers.values():
|
|
|
+# try:
|
|
|
+# i.create_topics(topic_reqs=(Topic("tag_sink_topic"),))
|
|
|
+# except Exception as e:
|
|
|
+# if re.search("41", str(e)):
|
|
|
+# print("该broker 不是 leader,交由下一个broker创建")
|
|
|
+# elif re.search("7", str(e)):
|
|
|
+# print("创建完成")
|
|
|
+# break
|
|
|
+# elif re.search("36", str(e)):
|
|
|
+# print("topic 已存在")
|
|
|
+# break
|
|
|
+# else:
|
|
|
+# raise e
|
|
|
+# print(client.topics)
|