spring-integration - spring集成kafka错误处理

Question

我们开始在一个项目中使用 spring-integration-kafka，但找不到任何关于消费者端错误处理建议方法的文档。
理想情况下，我们想要一个不阻塞线程的退避重试策略，因此，似乎无状态重试建议不是一种选择；但是，我找不到使用 Kafka 的有状态重试建议的示例。
关于我们应该如何使用 spring-integration-kafka 处理错误重试的任何建议？谢谢。

Answer 1

请分享有关此事的配置，并指出您想在哪里使用重试建议。

通常，任何inbound-channel-adapter将其消息发送到具有订阅者的某个频道，例如<service-activator>可以使用RequestHandlerRetryAdvice.

从另一侧可以<poller>根据<int-kafka:inbound-channel-adapter>需要进行配置StatefulRetryOperationsInterceptor。

Answer 2

由于您的要求是在重试期间不阻塞线程，因此您可以将失败的消息写入另一个主题，并让一些后台进程从这个“死信队列”中消耗并在一段时间后重新发布到原始主题。

使用rabbitmq，这可以通过适当配置死信交换和具有消息生存时间的死信队列来自动完成，但我认为kafka没有类似的东西，所以你必须自己动手。