使用kafka-python从外部连接Heroku Kafka实例无消息输出问题排查

看起来你的消费者配置有几个关键遗漏，导致无法正常接收消息。我帮你梳理一下问题点和修复方案：

1. 缺少SSL根证书（CA证书）配置 #

Heroku Kafka的SSL连接需要验证服务器端证书，你的代码里只指定了客户端证书和密钥，但没配置信任的根证书（也就是KAFKA_TRUSTED_CERT对应的文件）。这会导致SSL握手表面无报错，但实际无法建立有效的可信连接。

你需要在KafkaConsumer的参数里添加：

ssl_cafile="KAFKA_TRUSTED_CERT.txt"

2. 消费者偏移量重置策略问题 #

如果这是你第一次运行这个消费者，或者使用了默认的随机消费者组ID，默认的auto_offset_reset参数是latest——这意味着消费者只会接收它启动之后产生的消息。你之前用heroku kafka:topics:write写入的消息是在消费者启动前产生的，自然不会被消费到。

你需要显式设置auto_offset_reset='earliest'，让消费者从主题的起始位置开始消费。同时建议手动指定一个固定的group_id，避免默认随机组ID带来的不确定性。

3. 证书内容的换行符处理问题 #

从Heroku环境变量导出的证书内容，换行符通常是以\n转义形式存在的，直接写入文件的话，这些转义字符不会被解析成实际的换行，导致证书文件格式无效。你需要在写入时把\n替换成实际的换行符。

修改后的完整代码 #

import os
from kafka import KafkaConsumer

# 处理证书写入，替换转义的换行符为实际换行
for variable in ['KAFKA_TRUSTED_CERT', 'KAFKA_CLIENT_CERT', 'KAFKA_CLIENT_CERT_KEY']:
    cert_content = os.environ[variable].replace('\\n', '\n')
    with open(f'{variable}.txt', "w") as text_file:
        text_file.write(cert_content)

consumer = KafkaConsumer(
    'test-topic',
    bootstrap_servers=os.environ['KAFKA_URL'],
    security_protocol="SSL",
    ssl_cafile="KAFKA_TRUSTED_CERT.txt",
    ssl_certfile="KAFKA_CLIENT_CERT.txt",
    ssl_keyfile="KAFKA_CLIENT_CERT_KEY.txt",
    group_id="my-heroku-kafka-consumer-group",  # 显式指定固定组ID
    auto_offset_reset='earliest'  # 从主题最早的消息开始消费
)

print("Consumer started, waiting for messages...")
for msg in consumer:
    print(f"Received message: {msg.value.decode('utf-8')}")

验证步骤 #

1. 先停止之前运行的消费者，删除生成的三个证书文件（避免旧文件干扰）
2. 运行修改后的代码
3. 重新用heroku kafka:topics:write test-topic "this is a test 2"写入一条新消息
4. 此时消费者应该能立即接收到这条新消息，同时也会拉取之前的历史消息

如果还是有问题，你可以添加日志排查SSL连接细节：配置logging模块把kafka的日志级别设为DEBUG，这样能看到连接过程中的详细报错信息。

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