卡夫卡消耗无法处理的消息 - 以后如何重新处理破碎的消息？

Question

我们正在使用Spring Kafka实现卡夫卡消费者。正如我理解正确的话单信息的处理失败，有选项

• 不关心，只是ACK
• 不要使用一些重试处理一个RetryTemplate
• 如果连这不起作用做一些自定义故障处理使用RecoveryCallback

我想知道你的最佳做法是什么。我想到了简单的应用程序异常，例如DeserializationException（用于JSON格式的消息）或更长的本地存储停机时间等。这意味着需要一些额外的工作，如修补程序部署，以修复损坏的应用程序以便能够重新处理错误的消息。由于丢失消息（即不处理它们）对我们来说不是一种选择，唯一剩下的选择是IMO将有故障的消息存储在某个持久性存储中，例如， G。另一个“错误消息”例如Kafka话题，以便这些事件可以在稍后再次处理，并且不需要完全停止事件处理。

你如何处理这些情况？

Answer 1

一个例子是Spring Cloud Stream，它可以配置为将失败消息发布到另一个主题errors.foo;用户可以将其复制回原始主题以便稍后重试。

该逻辑在恢复回调中完成。

Answer 2

我们有一个用例，即使对于错误的消息我们也不能丢弃任何消息。因此，当我们遇到错误的消息时，我们将发送默认消息来代替该错误记录，同时将消息发送给失败的主题以便以后重试。