Python Kafka Consumer.poll报错：'dict'对象无'error'属性

核心原因：用错了Kafka Python客户端库 #

你遇到的问题是不同Kafka Python客户端的API差异，不是版本问题：

• 你看到的大量示例代码，用的是confluent-kafka客户端（Confluent维护的库），它的poll()方法返回单个Message对象，所以有error()、value()这类方法。
• 而你实际使用的是kafka-python库（对应原生Kafka+ZooKeeper的部署场景），这个库的consumer.poll()返回的是字典，结构为{TopicPartition: [Record, Record, ...]}，自然没有error()方法，直接调用就会报错。

适配kafka-python的正确写法 #

针对kafka-python库，你需要遍历poll返回的字典，逐个处理分区里的消息记录，示例代码如下：

import time
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'your_topic',
    bootstrap_servers='your_kafka_brokers',
    group_id='your_group_id',
    # 根据你的配置添加其他参数，比如auto_offset_reset等
)

while True:
    try:
        # poll返回：{TopicPartition: [Record]}，timeout单位为毫秒
        records_by_partition = consumer.poll(timeout_ms=10000)
        
        if not records_by_partition:
            print("No messages received, sleeping for 2 minutes")
            time.sleep(120)
            continue
        
        # 遍历每个分区的消息列表
        for partition, records in records_by_partition.items():
            for record in records:
                # 检查记录是否有错误（kafka-python中是属性而非方法）
                if record.error:
                    print(f"Consumer error in partition {partition}: {record.error}")
                    continue
                # 处理消息内容
                print(f"Received message: {record.value.decode('utf-8')}")
                
    except Exception as e:
        print(f"Unexpected error occurred: {str(e)}")
        # 可添加异常后的恢复逻辑，比如重新连接等

关键差异点总结 #

• confluent-kafka：poll()返回单个Message对象，用message.error()、message.value()调用
• kafka-python：poll()返回字典，需遍历分区和记录，用record.error（属性，不是方法）、record.value获取内容

内容的提问来源于stack exchange，提问作者mez63