如何在java的kafka中获取消费者组的Consumer Lag

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
props.put("group.id", "consumer-group");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

consumer.subscribe(Collections.singletonList("topic-name"));

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
Set<TopicPartition> partitions = consumer.assignment();

for (TopicPartition partition : partitions) {
    long offset = consumer.position(partition);
    // 处理消费者的偏移量
}

Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);

for (TopicPartition partition : partitions) {
    long offset = consumer.position(partition);
    long endOffset = endOffsets.get(partition);
    long lag = endOffset - offset;
    // 处理Consumer Lag
}