如何通过代码获取Kafka 2.1.1中指定Topic的log-end-offset？

当然有啦！不管你用Java、Python还是其他主流语言的Kafka客户端，都有对应的API可以获取log-end-offset这个参数。下面我分两种常用语言给你具体说说实现思路：

方法一：使用 AdminClient（推荐用于管理场景） #

Kafka的AdminClient是专门用于集群管理的客户端，能直接查询指定Topic的分区信息和对应log-end-offset：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class LogEndOffsetFetcher {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 配置Kafka连接信息
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "server1:9092");
        
        try (AdminClient adminClient = AdminClient.create(props)) {
            String targetTopic = "your-target-topic";
            // 获取目标Topic的详细描述
            TopicDescription topicDesc = adminClient.describeTopics(Collections.singleton(targetTopic))
                    .all().get().get(targetTopic);
            
            // 遍历每个分区，查询log-end-offset
            for (var partitionInfo : topicDesc.partitions()) {
                TopicPartition partition = new TopicPartition(targetTopic, partitionInfo.partition());
                // 查询该分区的最新偏移量
                Map<TopicPartition, Long> endOffsets = adminClient.listOffsets(Collections.singletonMap(partition, org.apache.kafka.clients.admin.ListOffsetsSpec.latest()))
                        .all().get();
                System.out.printf("分区 %d 的 log-end-offset: %d%n", partition.partition(), endOffsets.get(partition));
            }
        }
    }
}

方法二：使用 KafkaConsumer（适合已有消费逻辑的场景） #

如果你的代码里已经在用KafkaConsumer，也可以直接用它的endOffsets方法获取最新偏移量：

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;

public class ConsumerEndOffsetExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "server1:9092");
        props.put("group.id", "group1");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            String targetTopic = "your-target-topic";
            // 订阅目标Topic，获取分区元数据
            consumer.subscribe(Collections.singletonList(targetTopic));
            // 短暂等待consumer同步分区信息
            Thread.sleep(1000);
            
            // 获取所有分区的log-end-offset
            Map<TopicPartition, Long> endOffsets = consumer.endOffsets(consumer.assignment());
            for (var entry : endOffsets.entrySet()) {
                System.out.printf("分区 %d 的 log-end-offset: %d%n", entry.getKey().partition(), entry.getValue());
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

如果用Python开发，推荐使用confluent-kafka库，它的API简洁易用：

from confluent_kafka import Consumer, KafkaException, TopicPartition

def fetch_log_end_offset(bootstrap_servers, target_topic):
    conf = {
        'bootstrap.servers': bootstrap_servers,
        'group.id': 'group1',
        'auto.offset.reset': 'latest'
    }
    
    consumer = Consumer(conf)
    try:
        # 获取目标Topic的所有分区
        topic_meta = consumer.list_topics(target_topic)
        partitions = topic_meta.topics[target_topic].partitions.keys()
        topic_partitions = [TopicPartition(target_topic, p) for p in partitions]
        
        # 分配分区并查询水位偏移量（high就是log-end-offset）
        consumer.assign(topic_partitions)
        watermark_offsets = consumer.get_watermark_offsets(topic_partitions, timeout=5.0)
        
        for tp, (low_offset, high_offset) in zip(topic_partitions, watermark_offsets):
            print(f"分区 {tp.partition} 的 log-end-offset: {high_offset}")
    except KafkaException as e:
        print(f"查询失败: {str(e)}")
    finally:
        consumer.close()

# 调用示例
fetch_log_end_offset("server1:9092", "your-target-topic")

简单说下核心逻辑：log-end-offset本质就是每个分区的最新消息偏移量，所有客户端API的思路都是先拿到目标Topic的分区列表，再针对每个分区查询其最新偏移量，只是不同语言的API命名略有差异而已。

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