Search code examples
pythonapache-kafkakafka-consumer-apiconfluent-kafka-python

Confluent-Kafka Python - Describe consumer groups (to get the lag of each consumer group)


I want to get the details of the consumer group using confluent-kafka. The cli equivalent of that is `

./kafka-consumer-groups.sh --bootstrap-server XXXXXXXXX:9092 --describe --group my-group

My end goal is to get the value of lag from the output. Is there any method in confluent-kafka python API to get these details. There is a method in the java API but I couldn't find it in python API.

I tried using the describe_configs method in the adminClient API but it ended up throwing kafkaException with following details

This most likely occurs because of a request being malformed by the client library or the message was sent to an incompatible broker. See the broker logs for more details.


Solution

  • For now I have come up with the following solution. It's a work around to get the combined lag of a consumer group

     def get_lag(topic,numPartitions):
        diff = list()
        for i in range(numPartitions):
            topic_partition = TopicPartition(topic, partition=i)
            low, high = consumer.get_watermark_offsets(topic_partition)
            currentList = consumer.committed([topic_partition])
            current = currentList[0].offset
            diff.append(high-current)
        return sum(diff) # Combined Lag