`для .. в KafkaConsumer`: как выжить` eof`?

Question

for consumed_record in KafkaConsumer(config.kafka_topic,                                      
                                     bootstrap_servers=config.kafka_bootstrap_servers,
                                     group_id=config.kafka_group_id):
        __process_kafka_record(consumed_record)

Приведенный выше код работает нормально, кроме случаев, когда видно EOF. Затем for l oop завершается, но это не то поведение, которое мне бы хотелось. Я хочу, чтобы мой потребитель работал бесконечно, если не умирает весь процесс. Концептуально мне нужно что-то вроде if is_eof(): continue. Как правильно сделать это с пакетом python-kafka?

Answer 1

Похоже, вы используете kafka- python.

Правильный способ использовать записи - это обернуть это l oop в другое l oop, которое только останавливается, если вы хотите.

Проект kafka- python содержит несколько примеров , например:

consumer = KafkaConsumer(bootstrap_servers='localhost:9092')
consumer.subscribe(['my-topic'])

while not self.stop_event.is_set():
    for message in consumer:
        print(message)
        if self.stop_event.is_set():
            break

consumer.close()