Блог

Главная - Вопросы по программированию - Кафка прочитал тему -с-начала

Кафка прочитал тему -с-начала

#python #python-3.x #apache-kafka

#python #python-3.x #apache-kafka

Вопрос:

У меня есть кластер Kafka с 3 брокерами и 3 зоопарками. Темы создаются с использованием 1 раздела и 3 факторов репликации. Когда я пытаюсь прочитать из темы с помощью —partition — все в порядке.

 ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my_topic --offset 0 --partition 0

Но когда я пытаюсь прочитать с начала

 ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my_topic --from-beginning

нет никакого вывода

 Kafka version ./kafka-topics.sh --version 2.5.0 (Commit:66563e712b0b9f84)

Можете ли вы, пожалуйста, помочь мне понять, может быть, мне не хватает некоторых свойств конфигурации, и как я могу прочитать —с начала?

То же самое в моем коде Python:

 from kafka import TopicPartition, KafkaConsumer
consumer = KafkaConsumer(auto_offset_reset='smallest', bootstrap_servers='my_server:9092')
**consumer.assign([TopicPartition('my_topic', 0)])**
msg = next(consumer)
for msg in consumer:
    print (msg)

это работает только тогда, когда я показываю раздел 0

 **consumer.assign([TopicPartition('my_topic', 0)])**

Комментарии:

1. это действительно странно. Мне пришли в голову только две вещи: в вашем Pyhton-коде для вас bootstrap_servers установлено значение my_server , и в вашем вызове командной строки вы используете localhost . Вероятно, просто копирование-вставка, но просто хотел убедиться. Кроме того, в вашем коде, который вы используете для auto_offset_reset , установлено значение smallest . В последних версиях это значение изменилось на earliest и, видя, что ваша версия должна быть 2.5.0 , я тоже в замешательстве….

2. Находятся ли 3 брокера на одном компьютере? Все ли они здоровы? Каков статус темы о смещениях потребителя?