librdkafka生产者内存管理：何时释放消息至空闲列表？

核心原则：只有当librdkafka明确不再需要消息payload的所有权时，才能将其放回空闲列表。以下是具体的错误码区分规则：

可安全释放消息的场景（错误码/状态） #

• 成功状态：RD_KAFKA_RESP_ERR_NO_ERROR
  消息已成功投递到Kafka集群，librdkafka不再持有payload，可直接放回空闲列表。
• 永久错误类：这类错误意味着消息无法被投递，且librdkafka不会自动重试，包括：
  • RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE：消息大小超过Broker配置的上限
  • RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PARTITION：目标主题/分区不存在，且Broker未开启自动创建主题
  • RD_KAFKA_RESP_ERR_INVALID_MSG、RD_KAFKA_RESP_ERR_INVALID_KEY：消息或Key格式无效
  • RD_KAFKA_RESP_ERR_UNSUPPORTED_COMPRESSION：使用了Broker不支持的压缩格式
  • RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED、RD_KAFKA_RESP_ERR_CLUSTER_AUTHORIZATION_FAILED：无主题/集群操作权限
  • RD_KAFKA_RESP_ERR__MSG_TIMED_OUT：已达到最大重试次数，消息最终投递失败
• 客户端终止类：比如RD_KAFKA_RESP_ERR__DESTROY（生产者实例被销毁时的回调），此时所有未完成的消息所有权归还应用。

不可释放消息的场景（错误码） #

这类属于临时错误，librdkafka会自动重试投递，因此仍保留消息payload的所有权，绝对不能释放：

• 网络/连接类临时错误：RD_KAFKA_RESP_ERR__TRANSPORT、RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN、RD_KAFKA_RESP_ERR__SSL
• Broker端临时不可用/资源不足：RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS、RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND、RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE
• 单次请求超时：RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT（未达到最大重试次数时）
• 客户端内部队列满：RD_KAFKA_RESP_ERR__QUEUE_FULL（librdkafka会将消息重新放入内部队列等待重试）

额外判断技巧 #

如果不确定错误码是否属于可重试类型，可以直接调用librdkafka提供的rd_kafka_err_is_retriable()函数：返回false则说明错误不可重试，可安全释放消息；返回true则表示librdkafka会继续重试，不能释放。

示例代码片段：

void dr_msg_cb(rd_kafka_t *rk, const rd_kafka_message_t *msg, void *opaque) {
    MyFreeList *free_list = (MyFreeList*)msg->_private;
    if (msg->err) {
        if (!rd_kafka_err_is_retriable(msg->err)) {
            // 不可重试，释放消息到空闲列表
            free_list_push(free_list, (void*)msg->payload);
        }
        // 可重试错误，不做释放，librdkafka会自动处理
    } else {
        // 投递成功，释放消息
        free_list_push(free_list, (void*)msg->payload);
    }
}

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