我如何知道我的记录是否已使用Spring Kafka手动提交

Question

我想知道在spring Kafka中，当ackmode设置为manual时，commit是如何工作的。

下面是我在kafkaconfigContainerProperties.SetackMode(AbstractMessageListenerContainer.ackMode.Manual）中设置的属性；

侦听器代码

@KafkaListener(id="POC", topics = "TestTopic", group = "TestGroup")
    public void listen(ConsumerRecord<String,KafkaPayload> record, Acknowledgment acknowledgment) {
        countDownLatch.countDown();     
        acknowledgment.acknowledge();
}

我正在按照spring kafka文档执行确认操作，但这只意味着我的消息被标记为已发送但未消费（这是我的理解）。

>

在这种情况下，我是否应该调用commitsync()方法。如果是，我从哪里调用它，因为我需要获得对kafkaconsumer的引用。如果没有，它是如何在内部工作的，我可以跟踪它吗？

是否有commitid或某个返回的值？我的想法是知道一个特定的消费者记录是否被消费。我想存储该值用于内部跟踪。

这将真正帮助我区分有多少记录被消耗，有多少记录被挂起以及它们的状态。

Answer 1

我可以回答第一个问题。一切Rest对阿帕奇Kafka来说就像是一个故事。

由于我们不能从我们想要的地方执行commit，而只能从执行consumer.poll()的同一线程执行，因此我们将所有提交请求存储在内部KafkaMessageListenerContainer队列中，并在执行this.consumer.poll()之前在主使用者循环中查看该请求。

即使您使用manual_immediate，实际的consumer.commitsync()也是在与您的acknowledgment.acknowledg()不同的线程上执行的。

在consumer中的API如下所示：

public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets);

因此，没有任何commitid钩子要处理。

我认为在Apache Kafka中没有类似于not committed之类的概念。这些数据始终存在于主题日志中，并且直到特定的管理操作或压缩配置才会被删除。

我认为提交偏移量特性与消费者组的目的和根据JavaDocs我们拥有的：

* This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every
* rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API
* should not be used. The committed offset should be the next message your application will consume,
* i.e. lastProcessedMessageOffset + 1.

因此，当您的使用者死亡时，它将从其组的上次提交的偏移量重新启动。不同的组可以读取相同的数据，但从某个其他偏移。我想这绝对是为什么他们的API没有提供任何到实际状态的挂钩。根本就没有这样一个！