如何以编程方式获取 Java 中 Kafka 主题的最新偏移量答案

【问题标题】：How to programmatically get latest offset of a Kafka topic in Java如何以编程方式获取 Java 中 Kafka 主题的最新偏移量
【发布时间】：2020-08-04 10:15:43
【问题描述】：

这是我正在尝试的

Collection <TopicPartition> partitions = consumer.partitionsFor(topic).stream();

还有如何表明您已经结束或不再有消息可供使用。如果偏移量与当时经纪人的结束偏移量不匹配怎么办。

任何建议。

【问题讨论】：

不确定你问的是否清楚。您想获取 Kafka 主题中分区可用的最新偏移量吗？

标签： java apache-kafka kafka-consumer-api kafka-partition

【解决方案1】：

为了获得最新的偏移量，您可以使用命令行：

./bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
    --broker-list localhost:9092 \
    --topic topicName

或在 Java 中以编程方式：

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProperties)) {
    consumer.subscribe(Arrays.asList("topicName"));
        Set<TopicPartition> assignment;
        while ((assignment = consumer.assignment()).isEmpty()) {
            consumer.poll(Duration.ofMillis(500));
        }
        consumer.endOffsets(assignment).forEach((partition, offset) -> System.out.println(partition + ": " + offset));
}

现在，如果您想强制消费者从最新的偏移量开始消费，您可以使用以下属性：

props.put("auto.offset.reset", "latest"); 
// or props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

或强制它使用最新的偏移量使用

consumer.seekToEnd();

【讨论】：

您能否建议我们如何知道消费者队列是否已满？如果丢失消息，我们是否必须执行一些重试逻辑以在一定时间内继续获取消息？如果可以在 java 中以编程方式完成，你能发布吗？谢谢！
@Reddy 我不确定您是否可以实际检查是否没有留言。 Kafka 是关于实时数据流的，所以这个问题对我来说没有多大意义。
@ Giorgos Myrianthous Okay.Here is what I'm doing to get a record Optional> record = consumer.nextRecord(timeout); if (record.isPresent()) then return record.get() else null。但问题是在获取消息时看起来有问题，看到记录被遗漏了。是由于消费者的内部队列吗？虽然不确定关于消费者的内部队列是如何工作的？
@Reddy 提出一个新问题并与您遇到的新问题分享您的完整代码会容易得多。仅看一行代码很难判断。
当然我会添加新问题。