如何为confluent_kafka库的producer.poll()实现重试机制？

你当前的代码没有覆盖Kafka生产者的全链路错误场景，且存在几个会导致重试逻辑完全失效的写法问题，以下是可直接落地的实现方案：

首先明确核心逻辑差异：Kafka生产者是异步批量发送模型，和HTTP同步等待响应的逻辑不同，错误分为三类，不能直接套API非200重试的逻辑：

• 瞬时可重试错误：网络抖动、broker主从切换、分区临时限流、请求超时这类，客户端内置重试逻辑的处理效率远高于外层自定义重试
• 入队即时错误：本地发送队列打满、序列化失败这类错误会在调用produce()方法时直接抛出
• 终态不可重试错误：Topic不存在、权限校验失败、消息大小超过阈值、消息超时未送达这类错误，重试无意义，需要告警后人工介入

第一步：配置客户端内置自动重试 #

首先修改生产者初始化参数，优先用客户端原生能力处理瞬时错误，避免自定义外层重试带来的额外开销和重复消息问题：

from confluent_kafka import Producer, KafkaError
# 建议将Producer初始化为全局单例，不要每次发消息都新建实例
producer = Producer(
    {
        "bootstrap.servers": CONFLUENT_SERVER,
        "sasl.mechanism": "PLAIN",
        "security.protocol": "SASL_SSL",
        "sasl.username": CONFLUENT_USER,
        "sasl.password": secrets["confluent_secret"],
        "error_cb": error_cb,
        # 重试相关核心配置
        "retries": 5,  # 瞬时错误最大自动重试次数，可按业务容忍度调整为3-10
        "retry.backoff.ms": 1000,  # 每次重试间隔1秒，避免高频重试打满broker
        "acks": "all",  # 消息写入所有ISR副本才判定发送成功，避免丢消息
        "enable.idempotence": True,  # 开启幂等性，避免重试导致的消息重复问题
        "queue.buffering.max.messages": 100000,  # 本地发送队列最大长度，按业务峰值调整
        "message.timeout.ms": 30000,  # 单条消息最长发送超时时间，超过则判定为终态失败
    }
)

注意：开启幂等性配置时，客户端会自动将retries调整为不小于1、acks强制设为all，不会出现配置冲突。

第二步：处理produce调用时的即时错误 #

produce()方法是把消息放入本地发送队列就返回，遇到队列满等问题会直接抛异常，需要捕获后做短重试：

import time
import json

def send_kafka_message(payload, producer):
    """Parses payload into message and sets on kafka producer."""
    logger.info("Sending message on Kafka topic...")
    message_value = json.dumps(payload)
    target_partition = abs(hash(payload["_id"])) % 6
    max_queue_retry = 3

    for attempt in range(max_queue_retry):
        try:
            producer.produce(
                ORDER_TOPIC,
                value=message_value,
                callback=delivery_report,
                partition=target_partition,
            )
            # 触发后台IO事件处理，不要设为0阻塞过短
            producer.poll(1)
            break
        except BufferError:
            # 本地队列满，等待后台发送线程腾出队列空间后重试
            if attempt == max_queue_retry - 1:
                logger.error(f"Kafka本地队列已满，重试{max_queue_retry}次仍无法入队，消息id:{payload['_id']}")
                raise  # 重试耗尽后抛出异常，交给上游做降级处理
            time.sleep(0.1 * (attempt + 1))
            producer.poll(1)
        except Exception as e:
            # 序列化失败、参数非法等非可重试错误直接抛出
            logger.error(f"Kafka消息入队失败，非可重试错误，消息id:{payload['_id']}, 错误:{str(e)}")
            raise

必须注意：绝对不要每次发送消息都新建Producer实例。Producer是线程安全的长连接对象，反复初始化会导致未发送完成的消息丢失、TCP连接反复建立带来极大性能开销，进程生命周期内初始化一次即可，进程退出前必须调用producer.flush(30)等待所有队列内消息发送完成。

第三步：修改发送回调处理终态结果 #

客户端自动重试耗尽后，会通过delivery_report回调返回最终结果，需要在回调里区分错误码做二次重试或死信处理：

def delivery_report(err, msg):
    if err is None:
        # 消息发送成功
        logger.debug(f"消息发送成功，topic:{msg.topic()}, partition:{msg.partition()}, offset:{msg.offset()}")
        return
    
    # 消息发送终态失败
    retriable_error_codes = {
        KafkaError.TIMED_OUT,
        KafkaError.NOT_LEADER_FOR_PARTITION,
        KafkaError.LEADER_NOT_AVAILABLE,
        KafkaError.BROKER_NOT_AVAILABLE,
        KafkaError.REQUEST_TIMED_OUT,
    }
    if err.code() in retriable_error_codes:
        # 可重试错误，重新投递消息，注意添加重试次数计数，超过阈值后写入死信队列
        logger.warning(f"Kafka消息发送终态失败，触发二次重试，错误:{err}，topic:{msg.topic()}, partition:{msg.partition()}")
        # 此处加入重试计数逻辑，避免无限重试
        # retry_send_logic(msg.value(), msg.partition())
    else:
        # 不可重试错误，直接写入死信队列等待人工排查
        logger.error(f"Kafka消息发送失败，不可重试错误，错误:{err}, topic:{msg.topic()}, 消息内容:{msg.value()}")
        # 此处加入死信队列写入逻辑

常见踩坑点 #

• 不要只调用一次producer.poll()：confluent-kafka客户端依赖poll()方法处理后台IO事件、触发发送回调，如果发完消息立刻退出进程不调用poll()和flush()，消息可能根本没发出去
• 不要对所有错误盲目重试：权限错误、Topic不存在、消息超过大小限制这类问题重试没有任何意义，只会浪费资源
• 必须开启幂等性：不管是客户端内置重试还是自定义重试逻辑，都可能产生重复消息，幂等生产者可以保证单分区内消息不重复，不需要额外做去重逻辑

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