Как определить, существует ли тема kafka с помощью confluent-kafka-python

Question

Я использую пакет confluent-kafka-python для взаимодействия с сервером Kafka. Я могу успешно создать тему и подтолкнуть к ней события. Однако моя проблема заключается в том, что я запускаю несколько узлов (работающих в Docker), если второй экземпляр также пытается создать тему, я получаю сообщение об ошибке. Прежде чем создавать новую тему, мне нужно сначала проверить, существует ли эта тема.

from confluent_kafka.admin import AdminClient, NewTopic
kafka_admin = AdminClient({"bootstrap.servers": server})

# First check here if the topic already exists!
if not topic_exists(topic):  # <-- how to accomplish this?
    new_kafka_topic = NewTopic(topic, num_partitions=1, replication_factor=1)
    results = kafka_admin.create_topics([new_kafka_topic])

Спасибо за любую помощь!

Answer 1

У меня была такая же проблема, и я справился с ней следующим образом:

client = AdminClient({"bootstrap.servers": BROKER_URL})
topic_metadata = client.list_topics()
if topic_metadata.topics.get(self.topic_name) is None:
  self.create_topic()

Answer 2

The AdminClientкласс list_topicsметод позволяет передать имя темы, которую вы хотите проверить, поэтому вам не нужно читать полный (и потенциально большой) список существующих тем:

      list_topics([topic=None][, timeout=-1])

Документация: https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#id0