使用confluent-kafka Python库无法发送Kafka消息，代码挂起求助

问题背景 #

原本使用kafka库的KafkaProducer发送消息完全正常，代码如下：

from kafka import KafkaProducer
ssl_produce = KafkaProducer(bootstrap_servers='xxxx:xxx',
                     security_protocol='SASL_SSL',
                     sasl_mechanism='SCRAM-SHA-512',
                     sasl_plain_username='xxxx',
                     sasl_plain_password='xxxxx')

ssl_produce.send('xxx', key=b'xxx', value=message_data)

切换到confluent-kafka库后，代码出现挂起，消息无法发送，使用的主题、用户名和密码与之前一致，新代码如下：

from confluent_kafka import Producer, KafkaException
from confluent_kafka.admin import AdminClient, NewTopic
from confluent_kafka.serialization import StringSerializer

# Kafka broker properties
bootstrap_servers = 'xxx:xxx'
security_protocol = 'sasl_ssl'
sasl_mechanism = 'SCRAM-SHA-512'
sasl_username = 'xxxx'
sasl_password = 'xxx'

# Kafka topic
topic_name = 'xxxx'

# Create the Kafka producer configuration
producer_config = {
    'bootstrap.servers': bootstrap_servers,
    'security.protocol': security_protocol,
    'sasl.mechanism': sasl_mechanism,
    'sasl.username': sasl_username,
    'sasl.password': sasl_password,
    'auto.offset.reset': 'earliest',  # Read messages from the beginning of the topic
    'default.topic.config': {
        'auto.offset.reset': 'earliest'
    }
}

# Create the Kafka producer
producer = Producer(producer_config)

# Publish a message to Kafka
def publish_message(producer, topic, message):
    try:
        producer.produce(topic=topic, value=message)
        producer.flush()
        print('Message published successfully.')
    except KafkaException as e:
        print('Failed to publish message:', str(e))

admin_client = AdminClient(producer_config)


# Publish a sample message
message = 'Hello, Kafka!'
publish_message(producer, topic_name, message)

问题原因及解决方案 #

1. 混入消费者专属配置，导致客户端初始化异常 #

auto.offset.reset和default.topic.config都是消费者配置项，生产者完全不需要这些配置。无效配置会干扰librdkafka底层的初始化逻辑，进而导致发送阻塞。

2. SSL配置缺失（可能） #

kafka库可能默认使用系统CA证书，但confluent-kafka的librdkafka底层需要明确指定SSL信任配置：

• 如果使用公共CA证书，需指定ssl.ca.location为系统CA证书路径（比如/etc/ssl/certs/ca-certificates.crt）
• 如果是自签名证书且临时测试，可添加ssl.verify.mode: 'none'（生产环境禁止使用）

3. 多余的AdminClient实例抢占资源 #

代码中实例化的AdminClient使用了和生产者相同的配置，若AdminClient初始化时遇到连接问题，可能占用资源导致生产者无法正常工作，可先移除这部分代码测试。

修正后的代码 #

from confluent_kafka import Producer, KafkaException

# Kafka broker properties
bootstrap_servers = 'xxx:xxx'
security_protocol = 'SASL_SSL'
sasl_mechanism = 'SCRAM-SHA-512'
sasl_username = 'xxxx'
sasl_password = 'xxx'

# Kafka topic
topic_name = 'xxxx'

# 仅保留生产者必要配置
producer_config = {
    'bootstrap.servers': bootstrap_servers,
    'security.protocol': security_protocol,
    'sasl.mechanism': sasl_mechanism,
    'sasl.username': sasl_username,
    'sasl.password': sasl_password,
    # 可选：指定CA证书路径，根据实际情况添加
    # 'ssl.ca.location': '/etc/ssl/certs/ca-certificates.crt',
    # 可选：临时关闭SSL验证（生产环境禁用）
    # 'ssl.verify.mode': 'none'
}

producer = Producer(producer_config)

def publish_message(producer, topic, message):
    try:
        # 添加回调函数，直观查看发送状态
        def delivery_report(err, msg):
            if err:
                print(f'消息投递失败: {err}')
            else:
                print(f'消息投递成功: {msg.topic()} [{msg.partition()}]')
        
        producer.produce(topic=topic, value=message, callback=delivery_report)
        producer.flush()
        print('消息已提交')
    except KafkaException as e:
        print('发送消息失败:', str(e))

# 测试发送
message = 'Hello, Kafka!'
publish_message(producer, topic_name, message)

额外排查步骤 #

• 先移除AdminClient相关代码，单独测试生产者发送逻辑
• 通过delivery_report回调获取更详细的错误信息，定位具体问题
• 检查confluent-kafka版本与Kafka集群版本的兼容性（建议客户端版本不低于集群版本）

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