无法从所有分区中获取 Kafka 滞后

Question

有没有办法为分配到同一消费者组的所有消费者找到整个 kafka 滞后？

我只能得到分配分区的延迟。例如假设只有一个分区分配给消费者，下面的代码只会为该分区带来延迟。不适用于其他分区。

Set<TopicPartition> partitionSet = consumer.assignment();
        Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitionSet);
for(TopicPartition tp : partitionSet) {
            LOG.info("Topic:{}, EndOffset:{}, currentOffset:{}, LAG:{}",
                    tp.topic(), endOffsets.get(tp), consumer.position(tp), endOffsets.get(tp)-consumer.position(tp));
        }

基本上，希望找到所有分区的滞后总和，以了解一个主题的所有消费者（同一组）滞后多少。

另外，有没有类似于 kafka-consumer-groups 的 api 可用，并通过 bootstrap-server 和 group 作为参数来查找滞后？

./kafka-consumer-groups.sh --bootstrap-server --group --describe

Answer 1

以编程方式实现此目的的正确方法是使用 AdminClient API：

1. 使用listConsumerGroupOffsets() 获取组的已提交偏移量。

2. 获取日志结束偏移量。目前，您需要启动消费者并为步骤 1 中检索到的所有分区调用 endOffsets()。

   在 Kafka 2.5（预计 2020 年 2 月结束）中，有一个新的 AdminClient API 来检索日志结束偏移量listOffsets()，这将允许单独使用 AdminClient 来检索滞后。

3. 对于每个分区，从日志结束偏移量（步骤 2）中减去提交的偏移量（步骤 1）。

这基本上是kafka-consumer-groups.sh 在幕后所做的。因此，如果需要，请查看implementation of this tool。