Spring Kafka非阻塞重试机制中请求头重复累积问题咨询

针对你遇到的重试请求头在重试主题和死信队列中不断累积的问题，提供以下几种可行方案：

1. 自定义重试头管理逻辑，替换而非追加头信息 #

Spring Kafka默认的重试头处理会追加新字段，你可以通过自定义RetryTopicHeadersManager覆盖这一行为，每次重试时先移除旧的重试相关头，再添加最新的：

public class NonDuplicatingRetryHeadersManager extends DefaultRetryTopicHeadersManager {
    @Override
    public void addRetryHeaders(ProducerRecord<?, ?> producerRecord, RetryTopicHeaders headers) {
        // 移除已存在的重试头，避免累积
        producerRecord.headers().remove(RetryTopicHeaders.RETRY_ATTEMPTS);
        producerRecord.headers().remove(RetryTopicHeaders.BACKOFF_TIMESTAMP);
        producerRecord.headers().remove(RetryTopicHeaders.ORIGINAL_TIMESTAMP);
        // 添加最新的重试头信息
        super.addRetryHeaders(producerRecord, headers);
    }
}

然后在重试配置中指定这个自定义管理器：

@Bean
public RetryTopicConfiguration myRetryTopicConfig(KafkaTemplate<String, Object> kafkaTemplate) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .customHeadersManager(new NonDuplicatingRetryHeadersManager())
            .create(kafkaTemplate);
}

2. 在死信队列阶段清理重试头 #

如果只需要在DLT中移除累积的重试头，可以在死信消息发布前拦截处理，清理掉相关头字段：

@Bean
public DeadLetterPublishingRecoverer deadLetterPublishingRecoverer(KafkaTemplate<String, Object> kafkaTemplate) {
    return new DeadLetterPublishingRecoverer(kafkaTemplate, (record, ex) -> {
        // 清理所有重试相关请求头
        record.headers().remove(RetryTopicHeaders.RETRY_ATTEMPTS);
        record.headers().remove(RetryTopicHeaders.BACKOFF_TIMESTAMP);
        record.headers().remove(RetryTopicHeaders.ORIGINAL_TIMESTAMP);
        // 指定死信队列的主题和分区
        return new TopicPartition("your-dlt-topic", record.partition());
    });
}

3. 升级Spring Kafka版本 #

部分早期版本的Spring Kafka存在重试头累积的问题，升级到2.8.x及以上的稳定版本，框架默认已经优化了头处理逻辑，会更新而非追加重试相关头字段。

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