使用kafka-python可发消息，confluent-kafka却失败，求排查原因

你的confluent-kafka代码无法发送消息，核心原因是未处理发送过程中的错误反馈，且存在配置细节的隐性差异。以下是具体排查和修复方案：

1. 添加错误回调，定位失败根源 #

confluent-kafka的Producer.produce()默认不会主动抛出异常，发送失败的信息仅通过回调函数返回。添加回调是排查问题的关键步骤：

from confluent_kafka import Producer, KafkaError

def delivery_report(err, msg):
    """消息发送结果回调函数"""
    if err is not None:
        print(f"消息发送失败: {err}")
    else:
        print(f"消息成功发送到 {msg.topic()} [{msg.partition()}]")

producer = Producer({
    "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS,
    "security.protocol": 'SASL_SSL',
    "sasl.mechanism": 'SCRAM-SHA-512',
    "sasl.username": secret_creds[KAFKA_VAULT_SECRET_USER_KEY],
    "sasl.password": secret_creds[KAFKA_VAULT_SECRET_PASSWORD_KEY],
    "ssl.ca.location": kafka_certificate_path,
})

for message in [bytes('confluent_{}'.format(num), 'utf-8') for num in range(10)]:
    # 传递回调函数到produce方法
    producer.produce(topic=KAFKA_TOPIC_NAME, value=message, callback=delivery_report)
    # 调用poll触发回调处理（非阻塞，参数0表示立即返回）
    producer.poll(0)

# flush等待所有消息发送完成，并处理剩余回调
producer.flush()

通过回调输出的错误信息，你可以直接定位是认证失败、主题权限不足、SSL证书问题还是其他网络故障。

2. 检查配置的隐性差异 #

对比kafka-python的配置，confluent-kafka需要注意以下细节：

• 确认sasl.mechanism拼写完全匹配（你代码中SCRAM-SHA-512是正确的）
• 确保ssl.ca.location指向的CA证书文件路径正确、文件存在且进程有读取权限
• 部分集群可能需要显式设置ssl.endpoint.identification.algorithm为https（默认开启，若出现SSL握手失败可尝试添加）

3. 常见失败场景的修复建议 #

根据回调返回的错误，对应修复：

• 认证失败：检查sasl.username和sasl.password是否与kafka-python使用的完全一致，注意特殊字符、大小写是否匹配
• 主题不存在/无权限：确认主题名称正确，或联系集群管理员开通账号的主题写入权限
• SSL握手失败：更换有效CA证书，或检查集群的SSL配置要求

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