如何在Java中获取Kafka Topic消费滞后？求助listGroupOffsets方法缺失问题

你遇到的问题核心是用了旧版的Kafka AdminClient API（kafka.admin.AdminClient），这个类早已被官方废弃，而listGroupOffsets是新版客户端API（org.apache.kafka.clients.admin.AdminClient）才提供的方法。下面是完整的解决步骤和代码示例：

1. 确认依赖配置 #

首先确保你的项目依赖的是新版kafka-clients库，以Maven为例：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version> <!-- 替换为你实际使用的Kafka对应版本 -->
</dependency>

2. 使用新版AdminClient获取消费者偏移量 #

新版AdminClient是异步API，需要通过get()方法同步获取结果，以下是获取指定消费组、Topic分区偏移量的代码：

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

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

public class KafkaOffsetChecker {
    public static void main(String[] args) {
        // 配置AdminClient参数
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(AdminClientConfig.CLIENT_ID_CONFIG, "offset-checker-client");

        // 用try-with-resources自动关闭客户端
        try (AdminClient adminClient = AdminClient.create(props)) {
            String groupId = "groupID";
            TopicPartition targetPartition = new TopicPartition("topic", 0);

            // 查询指定消费组的偏移量
            ListGroupOffsetsResult groupOffsetsResult = adminClient.listGroupOffsets(groupId);
            Map<TopicPartition, Long> groupOffsets = groupOffsetsResult.partitionsToOffsetAndMetadata().get();

            // 获取目标分区的消费偏移量
            Long consumerOffset = groupOffsets.get(targetPartition);
            if (consumerOffset != null) {
                System.out.printf("消费组 %s 在分区 %s 的偏移量: %d%n", groupId, targetPartition, consumerOffset);
            } else {
                System.out.println("该消费组尚未消费过这个分区");
            }

        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

3. 完整计算消费滞后量 #

如果要真正查看消费滞后（即分区最新偏移量与消费者偏移量的差值），还需要获取Topic分区的最新偏移量，补充代码如下：

// 在上述try块中添加这段代码，获取最新偏移并计算滞后量
import org.apache.kafka.clients.admin.OffsetSpec;
import java.util.Collections;

// 获取分区最新偏移量
Map<TopicPartition, Long> latestOffsets = adminClient.listOffsets(
        Collections.singletonMap(targetPartition, OffsetSpec.latest())
).all().get();

Long currentLatestOffset = latestOffsets.get(targetPartition);
if (currentLatestOffset != null && consumerOffset != null) {
    long lag = currentLatestOffset - consumerOffset;
    System.out.printf("当前消费滞后量: %d%n", lag);
}

关键注意点 #

• 旧版kafka.admin.AdminClient已被标记为@Deprecated，官方强烈推荐使用org.apache.kafka.clients.admin.AdminClient。
• 新版API以异步操作为主，若不想阻塞等待结果，也可以使用回调函数处理响应。
• 务必保证Kafka服务端版本与客户端版本兼容，避免出现API不匹配的问题。

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