如何在Java中获取Kafka消费者组的Consumer Lag

我之前也碰到过这个问题，在Kafka 2.0.1版本里，AdminClient确实没有直接提供获取Consumer Lag的API——不像后来的版本（比如2.3+）新增了相关便捷方法。不过我们可以通过组合两个核心API手动计算Lag，逻辑和你用的kafka-consumer-groups命令完全一致：先拿到消费者组的当前消费偏移量，再拿到每个分区的最新消息偏移量，两者相减就是Lag。

实现步骤 #

1. 创建AdminClient实例并配置连接参数
2. 获取目标消费者组在各分区的当前消费偏移量
3. 获取对应分区的最新消息偏移量（即分区末尾的偏移量）
4. 计算每个分区的Lag：最新偏移量 - 当前消费偏移量

完整代码示例 #

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ConsumerGroupOffsetsResult;
import org.apache.kafka.clients.admin.ListOffsetsResult;
import org.apache.kafka.clients.admin.OffsetSpec;
import org.apache.kafka.common.TopicPartition;

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

public class ConsumerLagCalculator {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 配置AdminClient连接参数
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(AdminClientConfig.CLIENT_ID_CONFIG, "lag-calculator-client");
        
        try (AdminClient adminClient = AdminClient.create(props)) {
            String groupId = "MyGroupName";
            
            // 2. 获取消费者组的当前消费偏移量
            ConsumerGroupOffsetsResult groupOffsetsResult = adminClient.listConsumerGroupOffsets(groupId);
            Map<TopicPartition, org.apache.kafka.clients.consumer.OffsetAndMetadata> currentOffsets = groupOffsetsResult.partitionsToOffsetAndMetadata().get();
            
            if (currentOffsets.isEmpty()) {
                System.out.println("该消费者组没有记录任何偏移量，可能尚未开始消费或已重置偏移量");
                return;
            }
            
            // 3. 批量获取每个分区的最新偏移量
            Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
            for (TopicPartition tp : currentOffsets.keySet()) {
                offsetSpecs.put(tp, OffsetSpec.latest());
            }
            ListOffsetsResult latestOffsetsResult = adminClient.listOffsets(offsetSpecs);
            Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> latestOffsets = latestOffsetsResult.all().get();
            
            // 4. 计算并输出每个分区的Lag
            System.out.println("消费者组 " + groupId + " 的Consumer Lag情况：");
            for (Map.Entry<TopicPartition, org.apache.kafka.clients.consumer.OffsetAndMetadata> entry : currentOffsets.entrySet()) {
                TopicPartition tp = entry.getKey();
                long currentOffset = entry.getValue().offset();
                long latestOffset = latestOffsets.get(tp).offset();
                
                // 处理消费者未开始消费的特殊情况（currentOffset可能为-1）
                long lag = latestOffset - Math.max(currentOffset, 0);
                System.out.printf("Topic: %s, Partition: %d, Current Offset: %d, Latest Offset: %d, Lag: %d%n",
                        tp.topic(), tp.partition(), currentOffset, latestOffset, lag);
            }
        }
    }
}

关键说明 #

• 命令行的底层逻辑：kafka-consumer-groups --describe命令内部就是执行了上述流程，先拉取消费者组偏移量，再拉取分区最新偏移量，最后计算差值返回，和我们的代码逻辑完全对齐。
• 异常场景处理：代码中已经处理了消费者组无偏移量的情况，你还可以根据业务需求添加更多异常捕获（比如Kafka集群连接失败、消费者组不存在等）。
• 版本适配：这段代码完全适配kafka-clients 2.0.1版本，如果你后续升级到更高版本（比如2.3+），可以使用describeConsumerGroups结合ConsumerGroupDescription来简化操作，但在2.0.1版本中只能用这种手动计算的方式。

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