无法捕获confluent-kafka-python的REQTMOUT超时错误，该如何解决？

• confluent-kafka-python的produce()是异步方法，调用时仅将消息写入本地客户端队列，Broker侧返回的请求超时（REQTMOUT）、发送失败等错误由后台rdkafka线程处理，默认不会抛出到produce()的调用栈，现有try-except无法捕获这类异步错误。
• 你当前直接传入字典类型作为value，confluent-kafka默认不会自动序列化字典，本地序列化失败的错误也可能被遗漏。
• flush()方法存在返回值，代表超时时间内未成功发送的消息数量，现有代码未对该返回值做判断，无法识别发送超时场景。

1. 核心调整逻辑 #

• 生产者初始化时添加全局错误回调，捕获客户端级别的连接、请求超时错误。
• 调用produce()时添加投递回调，记录单条消息的发送结果。
• 对字典类型的value做序列化处理，比如转为JSON字符串。
• 调用flush()时指定超时时间，判断返回值识别未发送成功的消息。

2. 修复后代码示例 #

import json
import logging
from confluent_kafka import KafkaError, KafkaException, Producer

# 替换为你实际的生产者配置
producer_conf = {
    "bootstrap.servers": "你的kafka服务地址",
    "security.protocol": "ssl",
    # 全局错误回调，捕获REQTMOUT这类客户端级别错误
    "error_cb": lambda err: logging.error(f"Kafka全局错误: {err}"),
    # 可选配置：根据业务场景调整超时阈值
    "message.timeout.ms": 30000,
    "request.timeout.ms": 15000
}

producer = Producer(producer_conf)
# 可根据代码架构换成类属性、闭包变量存储错误状态
publish_error = False

def delivery_callback(err, msg):
    global publish_error
    if err:
        logging.error(f"消息发送失败: {err}")
        # 匹配超时类错误
        if err.code() in (KafkaError._MSG_TIMED_OUT, KafkaError.REQUEST_TIMED_OUT, KafkaError._ALL_BROKERS_DOWN):
            publish_error = True

def publish_data(topic, produced_epoch):
    global publish_error
    publish_error = False
    try:
        # 序列化字典为字节流，避免本地序列化错误
        value = json.dumps({"produced_epoch": produced_epoch}).encode("utf-8")
        producer.produce(
            topic=topic,
            value=value,
            on_delivery=delivery_callback
        )
        # flush指定超时时间，返回值为未成功发送的消息数
        remaining = producer.flush(timeout=30)
        if remaining > 0 or publish_error:
            logging.error(f"有{remaining}条消息未成功发送，或出现超时错误")
            return False
    except (KafkaError, KafkaException, Exception) as e:
        logging.error(f'发送消息到Kafka出错: {e}')
        return False
    return True

3. 补充说明 #

• 投递回调会在消息成功被Broker确认或者永久失败时触发，可以精准捕获REQTMOUT这类超时错误。
• 如果不希望用全局变量存储错误状态，可以用类封装生产者逻辑，用实例属性存储错误标记，或者使用闭包传递状态。
• 你可以根据业务需要调整message.timeout.ms和request.timeout.ms的配置值，避免不必要的超时。

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