Spring Kafka批量消费下如何配置CustomNonRetryableException仅重试1次后转DLT

核心思路 #

通过ExceptionClassifierBackOffManager为不同类型的异常配置差异化的退避（重试）策略，对CustomNonRetryableException单独设置重试次数，其他异常沿用原有3次尝试的逻辑。

具体实现步骤 #

1. 构建异常分类退避管理器

   • 为CustomNonRetryableException配置FixedBackOff(1000L, 1L)：即失败后重试1次，加上首次尝试共2次，之后直接发送到DLT。
   • 为其余所有异常设置默认的FixedBackOff(1000L, 2L)：保持原逻辑，重试2次，总共3次尝试后进入DLT。

2. 配置DefaultErrorHandler

   • 将分类退避管理器注入DefaultErrorHandler，同时绑定已配置好的DLT发送器（DeadLetterPublishingRecoverer），确保最终无法重试的消息正确路由到DLT主题。

3. 绑定到容器工厂

   • 将配置完成的DefaultErrorHandler设置到ConcurrentKafkaListenerContainerFactory中，开启批量消费并生效异常处理规则。

代码示例 #

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory,
            KafkaTemplate<String, String> kafkaTemplate) {

        // 配置DLT消息发送器，自动路由到原主题后缀为.dlt的DLT主题
        DeadLetterPublishingRecoverer dlpr = new DeadLetterPublishingRecoverer(kafkaTemplate,
                (record, ex) -> new TopicPartition(record.topic() + ".dlt", record.partition()));

        // 创建异常分类退避管理器
        ExceptionClassifierBackOffManager backOffManager = new ExceptionClassifierBackOffManager();
        // 为CustomNonRetryableException设置仅重试1次
        backOffManager.addClassifier(CustomNonRetryableException.class, new FixedBackOff(1000L, 1L));
        // 其他异常保持原3次尝试逻辑
        backOffManager.setDefaultBackOff(new FixedBackOff(1000L, 2L));

        // 初始化DefaultErrorHandler并绑定退避管理器和DLT恢复器
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(dlpr, backOffManager);

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setBatchListener(true); // 启用批量消费模式
        factory.setCommonErrorHandler(errorHandler); // 绑定错误处理器

        return factory;
    }
}

关键说明 #

• FixedBackOff的第二个参数是退避次数，总尝试次数 = 退避次数 + 1。例如FixedBackOff(1000L,1L)对应1次重试、2次总尝试。
• 如果需要对更多异常类型做差异化控制，可继续调用addClassifier方法添加对应规则。
• 批量消费场景下，若需针对单条消息处理异常而非整批重试，可额外配置BatchErrorHandler的子类型，但当前配置已满足重试次数控制的需求。

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