Как создавать темы, если они не существуют в Kafka динамически с использованием kafka-python

Question

Я довольно новичок в Python и только начинаю с Кафкой. Я использую библиотеку с именем python-kafka для связи с Кафкой. Теперь у меня есть требование, что мне нужно создавать темы динамически, однако, если оно существует, мне не нужно его создавать.

Из чтения документов я вижу, что могу использовать KafkaAdminClient для создания и удаления тем, однако я не могу найти ни одного, чтобы проверить, существует ли тема.

Answer 1

KafkaAdminClient не предоставляет метод для вывода списка тем, но вы можете получить список существующих тем, просто запросив метаданные кластера из KafkaClient .

Например, при этом будет напечатановсе темы в кластере:

from kafka.client import KafkaClient

client = KafkaClient(bootstrap_servers='localhost:9092')

future = client.cluster.request_update()
client.poll(future=future)

metadata = client.cluster
print(metadata.topics())