在@KafkaListener-方法中确认，而不会“丢失”消息

Question

我们目前基本上通过以下简化机制确认消息：

  @KafkaListener(topics = "someTopic")
  public void listen(final String message, final Acknowledgment ack) {
    try {
        processMessage(message);
        ack.acknowledge();
    } catch (final IOException e) {
        // do not acknowledge here since we can temporarily not process the message
    } 

基本上，每当我们暂时不能处理消息时（在IOExceptions的情况下），我们希望在以后的时间再次接收它。

但这不起作用，因为acknowledge假设同一分区内以前的所有消息都已成功处理。在我们的IOException案例中，失败的消息将被跳过，但可能会被同一分区上具有更高索引的不同消息确认。

我们对如何解决这个问题有一些想法，但这意味着需要一些棘手的解决方法，以避免在KafkaListener方法中直接调用acknowledge。我们的用例是非常具体的，还是更像SpringKafka用户假设的“默认”行为？

对于这类问题有没有一个SpringKafka的解决方案？或者你有没有办法“正确”解决这个问题？

Answer 1

这就是Kafka的工作方式；您可以启用“重试”以尝试在继续下一条消息之前通过IOException。

您可以配置一个错误句柄，以便将失败消息发布到另一个主题，以便稍后重播。或者，错误处理程序可以停止容器以阻止任何新的交付。

重新启动容器时，将重播消息。