如何使用Python版Kafka Describe Group API查询消费组consumer lag

存在可用的Python实现，主流的 Kafka Python 客户端都支持该能力，具体可以通过两类客户端实现：

• 第一种是 Confluent 官方提供的 confluent-kafka 客户端，原生内置了完整的Describe Group能力，配合 list_consumer_group_offsets API可以直接获取指定消费组的提交偏移量、消费组成员、分区分配等全部信息，再结合对应分区的最新消息偏移量，两者相减就能得到 consumer lag。
• 第二种是社区常用的 kafka-python 客户端，你可以通过 KafkaAdminClient 类的 describe_consumer_groups 方法实现完整的Describe Group功能，获取消费组的状态、成员、已提交偏移量等信息，再配合 KafkaConsumer 的 end_offsets 方法拉取分区最新偏移量，计算得到对应lag。

给你一个基于 confluent-kafka 的最简示例代码片段：

from confluent_kafka import Consumer, TopicPartition, KafkaException

# 配置基础参数
conf = {
    'bootstrap.servers': '替换为你的kafka broker地址',
    'group.id': '替换为目标消费组id'
}
consumer = Consumer(conf)

try:
    # 示例：查询指定topic下0号分区的消费lag
    tp = TopicPartition('替换为目标topic名称', 0)
    # 1. 获取消费组在该分区的已提交偏移量
    committed_tp = consumer.list_consumer_group_offsets([tp])[0]
    committed_offset = committed_tp.offset
    # 2. 获取该分区的最新消息偏移量
    _, latest_offset = consumer.get_watermark_offsets(tp)
    # 3. 计算lag
    lag = latest_offset - committed_offset if committed_offset >=0 else latest_offset
    print(f"目标分区消费lag为: {lag}")
except KafkaException as e:
    print(f"查询失败: {e}")
finally:
    consumer.close()

注意：如果需要批量查询集群全量消费组的lag，你可以先通过AdminClient拉取集群内所有消费组列表，再遍历执行上述查询逻辑即可。

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