KafkaProducer发送事件失败崩溃，请求调试指引

问题描述 #

使用kafka-python的KafkaProducer无法向Kafka Broker发送消息，日志显示连接正在关闭。已排除Broker、Topic及权限问题（confluent-kafka的Producer可正常运行），程序执行producer.send(...)后直接终止，需调试方向。

日志信息 #

INFO:kafka.conn:<BrokerConnection node_id=bootstrap-0 host=bootstrap-server:443 <connected> [IPv4 ('xx.xxx.xxx.xxx', 443)]>: Closing connection.

问题代码 #

from dotenv import load_dotenv
from pathlib import Path
from kafka.errors import KafkaError
from kafka import KafkaProducer

dotenv_path = Path("./local.env")
load_dotenv(dotenv_path=dotenv_path)

import json
import logging
import os

import sys
logging.basicConfig(stream=sys.stderr, level=logging.INFO)

def main(): # 原代码此处缺少冒号，属于语法错误，需修正
    try:
        bootstrap_server = os.environ['KAFKA_SERVERS']
        topic = os.environ['KAFKA_TOPIC']
        ssl_certificate_location = os.environ['SSL_CERTIFICATE_LOCATION']
        ssl_key_location = os.environ['SSL_KEY_LOCATION']
        ssl_ca_location = os.environ['SSL_CA_LOCATION']

        producer = KafkaProducer(bootstrap_servers=bootstrap_server,
                                 security_protocol='SSL',
                                 ssl_check_hostname=True,
                                 ssl_cafile=ssl_ca_location,
                                 ssl_certfile=ssl_certificate_location,
                                 ssl_keyfile=ssl_key_location)

        for i in range(5):
            d = {'val': i}
            print("Iteration: ", i)
            print("Is connected to bootstrap: ", producer.bootstrap_connected())
            # KafkaProducer的value参数需要字节类型，原代码中json.dumps返回字符串，需编码
            future_record_meta_data = producer.send(topic, value=json.dumps(d).encode('utf-8'))
           
            try:
                record_metadata = future_record_meta_data.get(timeout=None)
                print(record_metadata)
            except KafkaError as e:
                print("Unable to send message [%s]" % e)

    except Exception as e:
        logging.debug("Exception ", e)
        print(e)
   
if __name__ == '__main__':
    main()

注1：原代码中def main()缺少冒号，属于语法错误，运行时会直接抛出异常，需先修正；注2：KafkaProducer的value参数需要传入字节类型，原代码中json.dumps(d)返回字符串，需添加.encode('utf-8')转换为字节，否则会触发类型错误

正常运行的confluent-kafka代码片段 #

conf = {
           'bootstrap.servers': bootstrap_server,
           'security.protocol': 'SSL',
           'ssl.certificate.location': ssl_certificate_location,
           'ssl.key.location': ssl_key_location,
           'ssl.ca.location': ssl_ca_location
}
  
producer = Producer(**conf)
producer.produce(topic, payload_str)

版本信息 #

python version 3.11.0
Kafka broker version 2.5.0
kafka-python==2.0.2
confluent-kafka==2.1.1

调试方向建议 #

• 提升日志级别获取细节：将kafka相关日志级别设为DEBUG，查看SSL握手、连接建立的完整过程。修改日志配置：

  logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
  logging.getLogger('kafka.conn').setLevel(logging.DEBUG)
  

• 对齐SSL参数配置：
  • 若SSL密钥文件有密码，kafka-python需显式设置ssl_password参数，confluent-kafka可能通过其他方式处理
  • 临时将ssl_check_hostname设为False，测试是否为主机名验证导致连接关闭
• 检查bootstrap_servers格式：kafka-python的bootstrap_servers若为多地址，需确保是列表格式（如bootstrap_server.split(',')），confluent-kafka可直接处理逗号分隔的字符串
• 显式处理异步发送：在循环结束后添加producer.flush()，确保所有消息都被提交，避免程序提前终止
• 验证Python版本兼容性：kafka-python 2.0.2对Python 3.11的支持可能存在问题，尝试降级到Python 3.10或升级kafka-python到最新稳定版
• 确认SSL文件权限与路径：确保进程能读取SSL证书、密钥和CA文件，尽量使用绝对路径避免工作目录问题
• 指定SSL协议版本：Kafka Broker 2.5.0默认可能使用TLSv1.2，可在kafka-python中显式设置ssl_protocol='TLSv1_2'，避免协议不匹配

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