无法再次接收未提交的 Kafka 消息答案

【问题标题】：Unable to receive non-committed Kafka messages again无法再次接收未提交的 Kafka 消息
【发布时间】：2023-04-09 08:49:01
【问题描述】：

我有这个使用 Spring 实现的 Kafka 消费者：

public class MyKafkaConsumer<T> implements AcknowledgingMessageListener<String, T> {
   private final ConcurrentMessageListenerContainer<String, T> container;

    public MyKafkaConsumer(ConsumerFactory<String, T> consumerFactoryry) {
       ContainerProperties prop = new ContainerProperties("mytopic");
       prop.setGroupId("mygroupid");
       prop.setMessageListener(this);
       prop.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
       this.container = new ConcurrentMessageListenerContainer<>(consumerFactory, prop);
    }

   @Override
   public void onMessage(ConsumerRecord<String, T> record, Acknowledgment acknowledgment) {
        // Just log the message, but don't acknowledge
   }
}

Kafka 配置了以下两个属性：

isolation.level = read_uncommitted
enable.auto.commit = false

由于我没有调用 acknowledgement.acknowledge()，我希望偏移量不会被提交。我也希望多次收到相同的消息，但这不会发生。当我重新启动整个应用程序时，它只会碰巧收到相同的消息。这是为什么？我不应该多次收到相同的消息吗？

我确定没有提交偏移量，因为我在重新启动应用程序时再次收到该消息。那么，还有什么情况呢？

我注意到，如果我在“onMessage”方法中抛出异常，那么我实际上会多次收到相同的消息。在这种情况下有什么区别？在这两种情况下，偏移量不是以相同的方式提交吗？

【问题讨论】：

相当确定行为内置于消费者中，基于此，请查看“偏移量和消费者位置”部分kafka.apache.org/22/javadoc/org/apache/kafka/clients/consumer/…
这是有道理的......但是当我抛出异常时有什么不同？因为在那种情况下，我一直收到相同的消息
很可能是提供给您的 Consumer 类在调用 onMessage 后会更新 position 值，因此在 onMessage 中未捕获的异常的情况下，内部位置不会不会更新。（通常我会查看代码来确认这一点，但还没有找到正确的来源）。如果您只想接收一次消息，即使您在处理过程中遇到异常，您也可以在 onMessage 内添加一个不会重新抛出的 try/catch。在理想情况下，您会将违规消息扔到另一个队列中以进行手动处理，或者将其记录在某个地方。

标签： java spring apache-kafka

【解决方案1】：

默认是BATCH，记录处理完成后提交偏移量，这就是你不会再次收到消息的原因

BATCH：当poll()返回的所有记录都处理完后提交偏移量

提交偏移量为提交偏移量提供了几个选项。如果 enable.auto.commit 消费者属性为 true，Kafka 会根据其配置自动提交偏移量。如果为 false，则容器支持多个 AckMode 设置（在下一个列表中描述）。默认的 AckMode 是 BATCH。从 2.3 版开始，框架将 enable.auto.commit 设置为 false，除非在配置中明确设置。以前，如果未设置属性，则使用 Kafka 默认值 (true)。

消费者 poll() 方法返回一个或多个 ConsumerRecords。为每条记录调用 MessageListener。以下列表描述了容器对每个 AckMode 采取的操作（未使用事务时）：

RECORD：在监听器处理完记录返回时提交偏移量。

BATCH：当 poll() 返回的所有记录都处理完后，提交偏移量。

TIME：当poll()返回的所有记录都被处理完后提交offset，只要超过了上次提交后的ackTime即可。

COUNT：当 poll() 返回的所有记录都被处理完后，提交偏移量，只要自上次提交后收到了 ackCount 记录。

COUNT_TIME：类似于TIME和COUNT，但是如果任一条件为真则执行提交。

MANUAL：消息侦听器负责确认（）确认。之后，应用与 BATCH 相同的语义。

MANUAL_IMMEDIATE：当监听器调用Acknowledgment.acknowledge()方法时立即提交偏移量

【讨论】：