kafka-consumer-monitor.py

import sys
import subprocess

if len(sys.argv) == 1:
    print('please specify a group-id')
    sys.exit(-1)

KAFKA_HOSTS="localhost:9092"
if len(sys.argv) > 2:
    KAFKA_HOSTS=sys.argv[2]

consumer_group_id = sys.argv[1]
cmd = "/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server %s --group %s --describe" % (KAFKA_HOSTS, consumer_group_id)

offset_sum = 0
lag_sum = 0

cmd_run = subprocess.run(cmd.split(), stdout=subprocess.PIPE)


for line in cmd_run.stdout.decode('utf-8').splitlines():
    if line.startswith(consumer_group_id):
        parts = line.split()
        #print(parts)
        offset_sum += int(parts[4])
        lag_sum += int(parts[5])

print('current offset:', offset_sum)
print('current lag:', lag_sum)