如何从 read_committed Kafka Consumer 获取最后提交的偏移量

Question

我正在使用事务性 KafkaProducer 向主题发送消息。这工作正常。我使用具有 read_committed 隔离级别的 KafkaConsumer，但我对 seek 和 seekToEnd 方法有疑问。根据文档， seek 和 seekToEnd 方法给了我 LSO（最后稳定偏移）。但这有点令人困惑。因为它总是给我相同的价值，即主题的结束。无论最后一个条目是提交（由生产者）还是中止事务的一部分。 例如，在我中止最后 5 次尝试插入 20_000 条消息后，消费者不应读取最后 100_000 条记录。但是在 seekToEnd 期间，它会移动到主题的末尾（包括 100_000 条消息）。但是 poll() 不会返回它们。

我正在寻找一种方法来检索上次提交的偏移量（即生产者最后一次成功提交的消息）。似乎没有合适的 API 方法。那我需要自己动手吗？

选项是返回并轮询直到没有更多记录被检索到，这将导致最后提交的消息。但我会假设 Kafka 提供了这种方法。

我们使用 Kafka 1.0.0。

Answer 1

要获取主题分区的最后提交偏移量，您可以使用KafkaConsumer.committed(TopicPartition partition) 函数。

TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
Long committedOffset = consumer.committed(topicPartition).offset();
System.out.println("last committed offset: " + committedOffset);  

Answer 2

KafkaConsumer 类有一些不错的方法，例如：partitionFor、begginingOffsets 和 endOffsets 以及 commited 和 position。

检查哪一个适合您的需求。尤其要仔细考虑所有 4 种与偏移相关的方法。 partitionFor 方法返回完整的元数据对象和其他信息，但对于丰富日志记录很有用。