在Python中指示group_id时，Kafka未收到消息

Question

我使用的是 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()

Answer 1

我终于解决了。

这是我的问题：omkafka配置文件partitions.numberattr 是1默认的。

我们将其更改为100需要的值并且它开始工作了！我希望它能帮助你