Confluent Kafka：非压缩主题获取最早可用偏移量的替代方法

针对非压缩（non-compacted）主题，当get_watermark_offsets返回的低水位（low watermark）为0无法满足需求时，可以通过以下几种方式获取实际可消费的最早偏移量：

1. 使用Kafka AdminClient查询日志起始偏移量 #

通过AdminClient的listOffsets方法，指定OffsetSpec.earliest()来获取每个分区的最早可用偏移量，这是最直接的方式，它会返回当前主题分区中实际存在的最早消息偏移量（即未被保留策略删除的起始位置）。

示例代码（Java）：

AdminClient adminClient = AdminClient.create(props);
Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
TopicPartition tp = new TopicPartition("your_topic", 0);
offsetSpecs.put(tp, OffsetSpec.earliest());

Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> results = adminClient.listOffsets(offsetSpecs).all().get();
long earliestAvailableOffset = results.get(tp).offset();

2. 通过消费者API手动定位 #

如果你已经有一个消费者实例，可以直接调用seekToBeginning方法将消费者定位到分区的最早可用位置，之后再通过position方法获取该偏移量。这种方法不需要额外的AdminClient，适合在消费流程中直接获取。

示例代码（Java）：

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
TopicPartition tp = new TopicPartition("your_topic", 0);
consumer.assign(Collections.singletonList(tp));
consumer.seekToBeginning(Collections.singletonList(tp));
long earliestAvailableOffset = consumer.position(tp);

3. 查看Kafka Broker日志或使用命令行工具 #

• 命令行可以用kafka-run-class.sh执行kafka.tools.GetOffsetShell，指定--time -2（代表最早可用偏移量）：

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list <broker_host:port> --topic your_topic --time -2

该命令会返回每个分区的最早可用偏移量，结果格式为your_topic:0:12345（12345即为目标偏移量）。

需要注意的是，以上方法返回的偏移量都是基于当前Broker上实际存储的消息位置，会自动考虑消息保留时长的限制，不会返回已经被删除的消息偏移量。

内容的提问来源于stack exchange，提问作者rpd