spring-kafka - 何时使用 RecoveryCallback 与 KafkaListenerErrorHandler

Question

我想了解什么时候应该使用 org.springframework.retry.RecoveryCallback 和 org.springframework.kafka.listener.KafkaListenerErrorHandler？

截至今天，我正在使用一个类(实现 org.springframework.retry.RecoveryCallback)来记录错误消息并将消息发送到 DLT 并且它正在工作。为了向 DLT 发送消息，我使用了 Spring KafkaTemplate，然后我遇到了 KafkaListenerErrorHandler 和 DeadLetterPublishingRecoverer。现在，您能否建议我，我应该如何使用 KafkaListenerErrorHandler 和 DeadLetterPublishingRecoverer？这可以代替 RecoveryCallback 吗？

这是我当前的 kafkaListenerContainerFactory 代码

@Bean
public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {

ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();

factory.setConsumerFactory(primaryConsumerFactory());
factory.setRetryTemplate(retryTemplate());
factory.setRecoveryCallback(recoveryCallback);
factory.getContainerProperties().setAckMode(AckMode.RECORD);
factory.setConcurrency(1);  
factory.getContainerProperties().setMissingTopicsFatal(false);
return factory;   }

Answer 1

如果它现在可以正常工作，为什么要改变它？

有几个层，您可以根据需要选择哪一层进行错误处理。

KafkaListenerErrorHandler将在重试中为每次传递尝试调用，因此您通常不会将它与重试一起使用。

重试 RecoveryCallback在重试用尽后调用(或者，如果您将异常归类为不可重试，则立即调用)。

ErrorHandler - 在容器中，如果任何监听器抛出异常就会被调用，而不仅仅是 @KafkaListener s。

使用最新版本的框架，您可以用 SeekToCurrentErrorHandler 完全替换监听器级别的重试。配置了 DeadLetterPublishingRecoverer和一个 BackOff .
DeadLetterPublishingRecoverer用于容器错误处理程序，因为它需要原始 ConsumerRecord<?, ?> .
KafkaListenerErrorHandler只能访问 spring-messaging Message<?>这是从 ConsumerRecord<?, ?> 转换而来的.