Kafka消费者无法消费生产者写入主题的数据问题排查

问题背景 #

• 现象：Kafka消费者无法消费生产者写入PEC5主题的数据，两端分别运行后控制台无任何输出
• 需求：生产者生成1-300的数字序列，每条消息包含主题、key及数字二进制值；消费者读取消息并仅输出value值

代码错误分析 #

1. 生产者代码错误：key格式不符合要求 #

原生产者代码中，key被设置为元组类型：

key = (str(i), 'utf-8')

KafkaProducer要求key参数必须是字节类型，而非元组。错误的key格式会导致消息发送异常，甚至无法成功写入Kafka主题。

2. 消费者代码错误：消息取值方式错误 #

原消费者代码中尝试遍历message.values：

for value in message.values:
    print(value)

KafkaConsumer返回的message对象是单个消息实例，不存在values属性，正确的value取值应为message.value（字节类型），需要解码为字符串后输出。

3. 兼容性优化：消费者偏移量参数 #

旧版本kafka-python中auto_offset_reset='smallest'可用，但新版本已统一使用'earliest'作为标准值，建议替换以保证跨版本兼容性。

修复后的代码 #

修复后的生产者代码 #

from kafka import KafkaProducer
import time

producer = KafkaProducer(bootstrap_servers='Cloudera02:9092')

for i in range(1, 300):
    value = bytes(str(i), 'utf-8')
    # 将key改为正确的字节类型
    key = bytes(str(i), 'utf-8')
    producer.send('PEC5', key=key, value=value)
    time.sleep(3) 
producer.flush()
producer.close()  # 新增关闭连接，确保所有消息完成发送

修复后的消费者代码 #

from kafka import KafkaConsumer

# 使用'earliest'替代'smallest'，保证版本兼容性
consumer = KafkaConsumer(
    'PEC5', 
    bootstrap_servers='Cloudera02:9092', 
    auto_offset_reset='earliest', 
    consumer_timeout_ms=10000,
    group_id='pec5-consumer-group'  # 新增消费者组ID，便于偏移量管理
)

for message in consumer:
    # 解码字节类型的value为字符串并输出
    print(message.value.decode('utf-8'))

consumer.close()

额外排查建议 #

• 确认Cloudera02:9092地址可正常访问，Kafka集群状态正常
• 检查PEC5主题是否已创建，可通过命令kafka-topics.sh --list --bootstrap-server Cloudera02:9092验证
• 确保生产者和消费者使用的Kafka版本与kafka-python库版本兼容
• 查看Kafka broker日志，排查是否有消息发送/消费的异常记录

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