DeadLetterPublishingRecoverer 如何与批量消费者一起使用？答案

【问题标题】：How does DeadLetterPublishingRecoverer work with batch consumer?DeadLetterPublishingRecoverer 如何与批量消费者一起使用？
【发布时间】：2022-01-19 22:49:46
【问题描述】：

我正在尝试了解 DeadLetterPublishingRecoverer 在批量消费者的情况下如何工作。我能看到的文档中的大多数示例都是单一记录消费者的示例。

我想为批量消费者实现重试逻辑。如果重试在某个阈值后也失败（比如说 3 次），那么 DeadLetterPublishingRecoverer 应该将消息发布到死信主题。顺序对我的用例来说并不重要。

我不确定是否：

DeadLetterPublishingRecoverer 会将失败的记录发布到 DLT，批处理中的其余记录将由消费者处理。

或者

DeadLetterPublishingRecoverer 会将失败的记录以及整个批次（包括好消息）发布到 DLT。

public ConcurrentKafkaListenerContainerFactory concurrentKafkaListenerContainerFactory(KafkaTemplate<String, String> kafkaTemplate) {
       
        ConcurrentKafkaListenerContainerFactory<String, String> concurrentKafkaListenerContainerFactory = new ConcurrentKafkaListenerContainerFactory<>();
    
        concurrentKafkaListenerContainerFactory.setBatchListener(true);
        
        DeadLetterPublishingRecoverer deadLetterPublishingRecoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        RecoveringBatchErrorHandler recoveringBatchErrorHandler = new RecoveringBatchErrorHandler(deadLetterPublishingRecoverer,new FixedBackOff(1000L, 3L));

        concurrentKafkaListenerContainerFactory.setBatchErrorHandler(recoveringBatchErrorHandler);

        return concurrentKafkaListenerContainerFactory;
    }

【问题讨论】：

标签： java spring-kafka

【解决方案1】：

对于RecoveringBatchErrorHandler（DefaultErrorHandler 2.8 及更高版本），监听器必须抛出一个BatchListenerFailedException；这告诉错误处理程序在批处理中发生错误的位置。

在此之前的记录将提交其偏移量；重试失败的记录（以及批次中的后续记录）；当重试次数用尽时，失败的记录被发送到 DLT，剩余的记录被重试。

见https://docs.spring.io/spring-kafka/docs/current/reference/html/#default-eh

和

https://docs.spring.io/spring-kafka/docs/2.7.x/reference/html/#recovering-batch-eh

适用于 2.8 之前的版本。

【讨论】：