如何用kafka-python成功生产消息？解决元数据更新超时认证问题

你的代码主要存在两个问题，修改后即可正常发送消息：

1. 安全协议参数设置错误
   Kafka中没有PLAIN这个合法的安全协议值，当需要用户名密码认证且不启用SSL时，正确的协议应为SASL_PLAINTEXT——它表示使用SASL认证但传输不加密。需将security_protocol='PLAIN'改为security_protocol='SASL_PLAINTEXT'。

2. 异步发送未做同步处理
   producer.send()是异步方法，直接调用后程序可能在消息发送完成前就退出，导致元数据更新超时。需要添加同步等待或flush操作，确保消息被提交。

修改后的完整代码：

from kafka import KafkaProducer
import json

KAFKA_TOPIC = 'TESTTOPIC'
KAFKA_BROKERS = 'localhost:9092'

producer = KafkaProducer(
    bootstrap_servers=[KAFKA_BROKERS],
    sasl_plain_username='USER',
    sasl_plain_password='PASS',
    sasl_mechanism='PLAIN',
    security_protocol='SASL_PLAINTEXT',  # 修正安全协议
    value_serializer=lambda m: json.dumps(m).encode(),
    api_version=(0,10,2)
)

info_as_json = {'foo': 'bar'}

# 发送消息并等待结果，避免异步导致的超时
future = producer.send(KAFKA_TOPIC, info_as_json)
future.get(timeout=10)  # 等待发送完成，超时时间可按需调整

# 或者使用flush确保消息被提交（二选一即可）
# producer.flush()

producer.close()

额外确认项：

• 确保Kafka集群已配置SASL PLAIN认证，且用户名密码正确
• 确认bootstrap_servers的地址、端口无误，网络可正常连通Kafka broker

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