如何获取同一消费者组全分区总Kafka Lag？是否有对应API？

一、用Java客户端获取全分区Lag总和 #

你原来的代码只能拿到当前消费者实例分配的分区Lag，核心原因是consumer.assignment()仅返回当前实例负责的分区。要统计整个消费者组的全分区Lag总和，关键是获取整个组的已提交位移（而非当前消费者的本地position），再结合目标topic所有分区的最新offset来计算。

具体实现步骤和代码示例如下：

1. 通过AdminClient查询消费者组的全局位移：Kafka消费者组的位移存储在broker的__consumer_offsets主题中，AdminClient提供了官方API可以直接拉取整个组的所有分区位移——这也是kafka-consumer-groups.sh脚本底层使用的逻辑。
2. 获取目标topic的所有分区：通过消费者客户端拿到topic的完整分区列表。
3. 计算每个分区的Lag并累加总和：用每个分区的最新offset减去组的已提交位移，最终得到全组的总滞后量。

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

public class ConsumerGroupTotalLagCalculator {
    public static void main(String[] args) {
        String bootstrapServers = "your-bootstrap-server:9092";
        String groupId = "your-consumer-group-id";
        String targetTopic = "your-target-topic";

        // 初始化消费者（用于获取分区最新offset，也可替换为AdminClient实现）
        Properties consumerProps = new Properties();
        consumerProps.put("bootstrap.servers", bootstrapServers);
        consumerProps.put("group.id", groupId);
        consumerProps.put("key.deserializer", StringDeserializer.class.getName());
        consumerProps.put("value.deserializer", StringDeserializer.class.getName());

        try (Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
             AdminClient adminClient = AdminClient.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers))) {

            // 先确认消费者组处于稳定状态，避免Rebalancing期间数据不准确
            DescribeConsumerGroupsResult groupDescResult = adminClient.describeConsumerGroups(Collections.singletonList(groupId));
            ConsumerGroupDescription groupDesc = groupDescResult.all().get().values().iterator().next();
            if (!ConsumerGroupState.STABLE.equals(groupDesc.state())) {
                System.out.printf("Consumer group %s is in %s state, please wait for rebalance to complete%n", groupId, groupDesc.state());
                return;
            }

            // 获取整个消费者组的已提交位移
            ListConsumerGroupOffsetsResult offsetsResult = adminClient.listConsumerGroupOffsets(groupId);
            Map<TopicPartition, OffsetAndMetadata> groupGlobalOffsets = offsetsResult.partitionsToOffsetAndMetadata().get();

            // 获取目标topic的所有分区
            List<PartitionInfo> partitionInfos = consumer.partitionsFor(targetTopic);
            Set<TopicPartition> allTopicPartitions = partitionInfos.stream()
                    .map(pi -> new TopicPartition(pi.topic(), pi.partition()))
                    .collect(Collectors.toSet());

            // 获取所有分区的最新offset（end offset）
            Map<TopicPartition, Long> endOffsets = consumer.endOffsets(allTopicPartitions);

            // 计算每个分区的Lag并累加总Lag
            long totalGroupLag = 0;
            for (TopicPartition tp : allTopicPartitions) {
                // 处理分区未被消费过的情况，默认位移为0
                long committedOffset = groupGlobalOffsets.getOrDefault(tp, new OffsetAndMetadata(0L)).offset();
                long latestOffset = endOffsets.get(tp);
                // 避免负Lag（比如位移过期被清理的场景）
                long partitionLag = Math.max(latestOffset - committedOffset, 0);
                totalGroupLag += partitionLag;

                System.out.printf("Topic: %-20s Partition: %-3d CommittedOffset: %-10d LatestOffset: %-10d Lag: %-10d%n",
                        tp.topic(), tp.partition(), committedOffset, latestOffset, partitionLag);
            }

            System.out.printf("%nTotal Lag for consumer group '%s' on topic '%s': %d%n", groupId, targetTopic, totalGroupLag);

        } catch (InterruptedException | ExecutionException e) {
            System.err.println("Failed to calculate consumer group lag: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

二、有没有类似kafka-consumer-groups.sh的API？ #

当然有！kafka-consumer-groups.sh的--describe功能完全基于Kafka AdminClient API实现，上面的代码其实就是在模拟这个脚本的核心逻辑。

如果需要在程序中集成查询能力，直接使用AdminClient是最官方、最灵活的方式，它支持kafka-consumer-groups.sh的所有核心功能：

• 查询消费者组的状态（稳定/重平衡中/死亡等）
• 获取组的全部分区位移
• 重置消费者组位移（如果业务需要）

如果用Spring Kafka等框架，也有封装好的工具类可以简化开发，但底层依然依赖AdminClient。

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