Spring Cloud Kafka 失败时会无限重试答案

【问题标题】：Spring cloud Kafka does infinite retry when it failsSpring Cloud Kafka 失败时会无限重试
【发布时间】：2022-04-18 18:47:15
【问题描述】：

目前，我遇到一个问题，其中一个消费者函数引发错误，导致 Kafka 一次又一次地重试记录。

@Bean
public Consumer<List<RuleEngineSubject>> processCohort() {
    return personDtoList -> {
        
        for(RuleEngineSubject subject : personDtoList)
            processSubject(subject);

    };
}

这是消费者processSubject 抛出一个导致失败的自定义错误。

processCohort-in-0:
  destination: internal-process-cohort
  consumer:
    max-attempts: 1
    batch-mode: true
    concurrency: 10
  group: process-cohort-group

以上是我的卡夫卡活页夹。

目前，我正在尝试重试 2 次，然后发送到死信队列，但我没有成功，并且不确定哪种方法是正确的。

我已尝试实现一个自定义处理程序，该处理程序将在失败时处理错误但不会再次重试，我不确定如何发送到死信队列

   @Bean
ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer() {

    return (container, dest, group) -> {
        if (group.equals("process-cohort-group")) {
            container.setBatchErrorHandler(new BatchErrorHandler() {
                @Override
                public void handle(Exception thrownException, ConsumerRecords<?, ?> data) {
                    System.out.println(data.records(dest).iterator().);
                    data.records(dest).forEach(r -> {
                        System.out.println(r.value());
                    });
                    System.out.println("failed payload='{}'" + thrownException.getLocalizedMessage());
                }
            });
        }
      
    };

}

这会停止无限重试，但不会发送死信队列。我能否获得有关如何重试两次然后发送死信队列的建议。据我了解，批处理侦听器在出现错误时不会如何恢复，有人可以帮助阐明这一点

【问题讨论】：

标签： apache-kafka spring-cloud spring-cloud-stream

【解决方案1】：

重试 15 次然后扔到 topicname.DLT 主题

@Bean
      public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setCommonErrorHandler(
            new DefaultErrorHandler(
                new DeadLetterPublishingRecoverer(kafkaTemplate()), kafkaBackOffPolicy()));
        factory.setConsumerFactory(kafkaConsumerFactory());
        return factory;
      }
    
@Bean
      public ExponentialBackOffWithMaxRetries kafkaBackOffPolicy() {
        var exponentialBackOff = new ExponentialBackOffWithMaxRetries(15);
        exponentialBackOff.setInitialInterval(Duration.ofMillis(500).toMillis());
        exponentialBackOff.setMultiplier(2);
        exponentialBackOff.setMaxInterval(Duration.ofSeconds(2).toMillis());
        return exponentialBackOff;
      }

【讨论】：

【解决方案2】：

你需要在监听容器中配置一个合适的错误处理器；您可以在绑定中禁用重试和 dlq，并改用 DeadLetterPublishingRecoverer。看答案Retry max 3 times when consuming batches in Spring Cloud Stream Kafka Binder

【讨论】：

引起：java.lang.IllegalStateException：只允许一个 LastSubscriberMessageHandler。实施时收到此错误您碰巧知道为什么吗？
可能不相关 - 使用 MCRE 提出新问题。