何时使用 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- 位于容器中，如果任何侦听器抛出异常（而不仅仅是@KafkaListeners），则调用它。

使用该框架的最新版本，您可以使用SeekToCurrentErrorHandler配置有 aDeadLetterPublishingRecoverer和 a 的监听器级别重试完全替换BackOff。

它DeadLetterPublishingRecoverer旨在用于容器错误处理程序，因为它需要原始ConsumerRecord<?, ?>.

唯一KafkaListenerErrorHandler可以访问Message<?>从ConsumerRecord<?, ?>.