Python使用KafkaConsumer读取topic消息无报错无返回问题求助

以下是代码中存在的错误和遗漏配置项：

• 缺失必填配置group_id：kafka-python消费者实例必须指定group_id参数，无该配置会导致消费者组协调异常，无法正常分配分区拉取消息。
• api_version配置错误：当前Kafka官方最高稳定版本为3.x系列，你配置的(20, 2, 1)属于不存在的版本号，会导致客户端和服务端版本协商失败，无法正常发起拉取请求。建议去掉该配置让客户端自动协商，或者填写和你部署的Kafka服务端一致的版本号。
• poll()方法返回值处理逻辑错误：consumer.poll()返回的是字典结构，格式为{TopicPartition实例: 分区对应的ConsumerRecord消息列表}，你当前直接对返回结果调用.value()、.headers()等方法是错误的，只有遍历到单个ConsumerRecord对象才能调用这些方法。
• 拉取超时和循环次数设置不合理：你设置poll的超时时间仅为0.05秒，且只循环10次，总等待时长不足1秒，消费者可能还没完成和broker的连接、分区分配等初始化动作就已经结束循环，返回空结果。

修正后的参考代码：

import json
from kafka import KafkaConsumer

def consumer():
    consumer = KafkaConsumer(
        'TOPIC_NAME',
        group_id='test_consumer_group_001', # 自定义消费组ID
        enable_auto_commit=False,
        value_deserializer=lambda m: json.loads(m.decode('ascii')),
        bootstrap_servers=['xx.x.xx.xx:port'],
        auto_offset_reset='earliest',
        consumer_timeout_ms=3000
    )

    for _ in range(20):
        msg_dict = consumer.poll(timeout_ms=500)
        if msg_dict:
            for partition, records in msg_dict.items():
                for record in records:
                    dat = {
                        'msg_value': record.value(),
                        'msg_headers': record.headers(),
                        'msg_key': record.key(),
                        'msg_partition': record.partition(),
                        'msg_topic': record.topic(),
                    }
                    print(dat)
        else:
            print('No Msg found')
    consumer.close()

补充排查项：
如果修改后仍无法读取消息，可以依次验证：

1. 客户端所在机器可以正常访问bootstrap_servers配置的地址和端口，无防火墙/安全组限制
2. Topic名称拼写完全正确，大小写匹配
3. 消息序列化规则和代码反序列化逻辑一致，如果消息不是ascii编码的json，可以先去掉value_deserializer配置，直接打印原始字节内容验证

内容的提问来源于stack exchange，提问作者kaviya .P