spring集成kafka错误处理答案

【问题标题】：spring integration kafka error handlingspring集成kafka错误处理
【发布时间】：2015-05-24 23:09:59
【问题描述】：

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

【问题讨论】：

即使有状态的重试也会阻塞线程——唯一的区别是异常是否被抛出回容器。对 Kafka 使用有状态重试没有多大意义，因为没有竞争消费者的概念。
感谢您的回答。你建议如何重试呢？例如，如果我正在使用队列，在 3 次非退避重试后，我可以将其扔回队列中，Kafka 中是否有类似的东西，例如，重试后，如果仍然失败，不提交位置？

标签： spring-integration apache-kafka spring-retry

【解决方案1】：

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

对于 rabbitmq，这可以是 done automatically with appropriate configuration of a dead letter exchange, and a dead letter queue with message time to live，但我认为 kafka 没有类似的东西，所以你必须自己动手。

【讨论】：

是的，我也不知道 Kafka 是怎么做到的，我确实使用了 ActiveMQ 的 DLQ，我自己会为 Kafka 模拟一个。谢谢。

【解决方案2】：

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

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

另一方面，<int-kafka:inbound-channel-adapter> 上的<poller> 可以根据需要配置为StatefulRetryOperationsInterceptor。

【讨论】：

根据上面 Gary 的回答，似乎这不是如何配置有状态或无状态建议的问题，而是我应该如何使用 Kafka 处理重试失败的问题。无论如何感谢您的提示。
您无法使用Hight-Level Consumer (<int-kakfa:inbound-channel-adapter>) 实现这一目标，因为它会在没有我们干预的情况下自动提交偏移量。对于<int-kafka:message-driven-channel-adapter> (Simple Consumer)，我们有一个开放的JIRA 票证，将在不久的将来解决。