Кафка не получает сообщения при указании group_id в Python - PullRequest
Новая
       66

Кафка не получает сообщения при указании group_id в Python

0 голосов
/ 21 сентября 2018

Я использую Kafka (kafka-python) версии 3.0.0-1.3.0.0.p0.40.Мне нужно настроить потребителя для темы «симуляция» в Python.Когда я не указываю group_id, т.е. group_id = None, он получает сообщения нормально.Однако, если я укажу group_id, он не получит никаких сообщений.

Вот мой код на Python: 

consumer = KafkaConsumer(bootstrap_servers='XXX.XXX.XXX.XXX:9092',
                         group_id = 'myTestGroupID', enable_auto_commit = True)
consumer.subscribe(['simulation'])
# not using assign method here as auto_commit is enabled
# partitions = [TopicPartition('simulation',num) for num in range(0,9)]
# consumer.assign([TopicPartition('simulation', partitions[0])])

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

Я попытался найти некоторые значения по умолчанию group_id в файлах свойств потребителя, я нашел один cloudera_mirrormaker, но ничего не изменилось.Мне нужно будет использовать несколько потребителей, поэтому важно, чтобы у меня был group_id, и они использовали один и тот же group_id.Во многих источниках я обнаружил, что group_id может быть любой строкой ...

Когда я запускаю потребителя для этой темы в консоли, он работает и получает сообщения 

./kafka-console-consumer.sh --bootstrap-server XXX.XXX.XXX.XXX:9092 --topic simulation --from-beginning --consumer-property group.id=myTestGroupID  --partition 0

, когда яЯ запускаю kafka-consumer-groups.sh, чтобы вывести список всех доступных групп, он пуст.

Если у кого-то есть идея, почему он застрял в Python, это было бы очень признательно.Большое спасибо

Вот код для производителя (я сократил его для простоты, так как в этом случае это не меняет проблему) 

from kafka import KafkaProducer
class Producer(threading.Thread):
    ...
    def run(self):
        producer = KafkaProducer(bootstrap_servers='XXX.XXX.XXX.XXX:9092')
        while not self.stop_event.is_set():
            string = 'test %s' %time.time()
            producer.send('simulation', string.encode())
            time.sleep(0.5)
        producer.close()

1 Ответ

0 голосов
/ 21 сентября 2018

Вопрос : Кафка не получает сообщения при указании group_id

Попробуйте, передайте 'theme' на KafkaConsumer, например, какв Документация : 

# join a consumer group for dynamic partition assignment and offset commits
from kafka import KafkaConsumer
consumer = KafkaConsumer('simulation', group_id='myTestGroupID')
for msg in consumer:
    print (msg)

Документация : KafkaConsumer ясно о типе group-id:

group_id (str или None) - имя группы потребителей, к которой нужно присоединиться для динамического назначения раздела (если оно включено) и которое используется для получения и фиксации смещений.Если нет, автоматическое назначение раздела (через координатора группы) и фиксация смещения отключены.По умолчанию: Нет

...