Monitor Kafka Consumer Lag using Python and Prometheus

Darian Vance here. Let’s talk about something that used to be a real thorn in my side: Kafka consumer lag. I remember my early days of managing our Kafka clusters, where “monitoring lag” meant SSH-ing into a box, running a command-line tool, and trying to parse the output. I probably wasted a couple of hours a week just manually checking things, especially during high-traffic events. It was reactive, inefficient, and frankly, a terrible use of my time.

That’s when I decided to automate it. By building a simple Python script to push these metrics to Prometheus, I turned a manual chore into a proactive, automated alerting system. Now, I get a Slack notification if a consumer falls behind, long before it becomes a production incident. This guide will walk you through setting up the exact same system. It’s a small investment of time for a massive gain in visibility and peace of mind.

Prerequisites

Before we dive in, make sure you have the following ready to go:

• Python 3.6 or newer.
• Access to a running Kafka cluster and you know the bootstrap server addresses.
• A running Prometheus server.
• A Prometheus Pushgateway instance. We use this because our script will be a short-lived job, not a persistent service that Prometheus can scrape directly.
• A basic understanding of Kafka concepts like topics, partitions, and consumer groups.

The Step-by-Step Guide

Step 1: Prepare Your Python Environment

First things first, let’s get our project space ready. I won’t walk you through the standard virtual environment setup (`venv` or `conda`), as I’m sure you have your own preferred workflow for that. The important part is to get the necessary libraries installed. In your activated environment, you’ll need to install two key packages using pip: `kafka-python` for interacting with Kafka, and `prometheus_client` for sending our metrics to the Pushgateway.

Step 2: The Python Script – Calculating Lag

This is where the magic happens. The core logic is surprisingly simple: for a given topic partition, we find the latest message offset (the “high-water mark”) and then find the last offset committed by our consumer group. The difference is the lag.

Let’s build the script. We’ll start by connecting to Kafka and defining the consumer groups and topics we want to monitor.

import os
from kafka import KafkaConsumer, TopicPartition
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway

# --- Configuration ---
# In a real setup, pull these from a config file or environment variables
KAFKA_BOOTSTRAP_SERVERS = ['kafka1:9092', 'kafka2:9092']
PUSHGATEWAY_ADDRESS = 'prometheus-pushgateway:9091'
# A dictionary mapping consumer groups to the topics they consume
CONSUMER_GROUPS_TO_MONITOR = {
    'payment-processor-group': ['payment-events-topic'],
    'notification-service-group': ['user-notification-topic', 'system-alert-topic']
}
JOB_NAME = 'kafka-lag-monitor'

def get_kafka_lag():
    """
    Connects to Kafka and calculates the consumer lag for specified groups and topics.
    """
    print("Starting Kafka lag check...")
    registry = CollectorRegistry()
    lag_gauge = Gauge(
        'kafka_consumer_lag_by_partition',
        'Current Kafka consumer lag per partition',
        ['consumer_group', 'topic', 'partition'],
        registry=registry
    )

    try:
        # We need a consumer to fetch metadata, but we won't actually consume messages.
        consumer = KafkaConsumer(
            bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
            client_id='lag-monitor-client',
            # Add security protocol settings here if needed, e.g., SASL_SSL
        )

        for group, topics in CONSUMER_GROUPS_TO_MONITOR.items():
            for topic in topics:
                partitions = consumer.partitions_for_topic(topic)
                if not partitions:
                    print(f"Warning: Could not find partitions for topic {topic}. Skipping.")
                    continue

                for partition in partitions:
                    tp = TopicPartition(topic, partition)
                    
                    # Get the last committed offset for the consumer group
                    committed_offset = consumer.committed(tp)
                    if committed_offset is None:
                        # If a group has never committed, we'll consider its offset 0.
                        # This avoids errors but means lag will equal the highwater mark.
                        committed_offset = 0

                    # Get the latest available offset (high-water mark)
                    end_offsets = consumer.end_offsets([tp])
                    latest_offset = end_offsets.get(tp, 0)

                    # Calculate lag
                    lag = latest_offset - committed_offset
                    # Ensure lag is never negative
                    lag = max(0, lag)
                    
                    print(f"Group: {group}, Topic: {topic}, Partition: {partition}, Lag: {lag}")
                    lag_gauge.labels(
                        consumer_group=group, 
                        topic=topic, 
                        partition=str(partition)
                    ).set(lag)
        
        # Push all collected metrics to the Pushgateway
        push_to_gateway(PUSHGATEWAY_ADDRESS, job=JOB_NAME, registry=registry)
        print("Successfully pushed metrics to Pushgateway.")

    except Exception as e:
        print(f"An error occurred: {e}")
        # In production, you'd want more robust error handling/logging here
        return 1 # Indicate failure
    
    finally:
        if 'consumer' in locals() and consumer:
            consumer.close()
    
    return 0 # Indicate success

if __name__ == '__main__':
    get_kafka_lag()

Pro Tip: Notice the labels in our Prometheus `Gauge`: `consumer_group`, `topic`, and `partition`. This level of detail is crucial. It allows you to pinpoint exactly which partition of which topic is falling behind, which is invaluable for debugging. Don’t skimp on labels!

Step 3: Scheduling the Script

This script isn’t a long-running service; it’s a point-in-time check. The easiest way to run it periodically is with a cron job. You can set it to run every minute, every five minutes, or whatever frequency makes sense for your use case.

A cron entry would look something like this. Remember to use the path to your Python executable and script.
`* * * * * python3 /path/to/your/kafka_lag_monitor.py`

Pro Tip: When I’m deploying this in a containerized environment like Kubernetes, I use a `CronJob` resource. It’s the cloud-native equivalent and manages the scheduling, retries, and logging beautifully. For a simple VM setup, standard cron is perfectly fine.

Step 4: Viewing the Metrics in Prometheus & Grafana

Once the script runs and pushes data, you can head over to your Prometheus UI. In the expression browser, you can query for your new metric: `kafka_consumer_lag_by_partition`.

The real power, however, comes from visualizing this in a dashboard like Grafana. You can build a graph that shows lag over time for a specific consumer group using a PromQL query like this:

`sum(kafka_consumer_lag_by_partition{job=”kafka-lag-monitor”, consumer_group=”payment-processor-group”}) by (topic)`

This query sums up the lag across all partitions for the `payment-processor-group` and shows you the total lag per topic. From there, you can set up alerting rules in Grafana or Alertmanager to notify you when the lag exceeds a certain threshold.

Common Pitfalls (Where I Usually Mess Up)

• Firewall/Network Issues: The most common problem I run into is the script not being able to reach the Kafka brokers or the Pushgateway. Always double-check your network security groups and firewalls. A simple `telnet` can save you a lot of time.
• Kafka Authentication: The example script uses a basic connection. In my production setups, Kafka is always secured with SASL or SSL. Forgetting to add the correct `security_protocol`, `sasl_mechanism`, and other auth parameters to the `KafkaConsumer` constructor will cause connection failures.
• Mismatched Job Labels: If you’re pushing metrics but they seem to disappear, check the `job` label. Prometheus uses the `job` and instance labels to uniquely identify a target. If you accidentally change the job name in your script, Prometheus will see it as a new, unrelated time series. Be consistent!

Conclusion

And there you have it. You’ve now moved from reactive, manual checks to a fully automated, proactive monitoring system for your Kafka consumers. This setup provides the data you need to build dashboards, configure alerts, and sleep better at night knowing you’ll be the first to know if a consumer starts to struggle. It’s a foundational piece of any robust data streaming architecture. Happy monitoring

Darian Vance

Lead Cloud Architect & DevOps Strategist

With over 12 years in system architecture and automation, Darian specializes in simplifying complex cloud infrastructures. An advocate for open-source solutions, he founded TechResolve to provide engineers with actionable, battle-tested troubleshooting guides and robust software alternatives.

🤖 Frequently Asked Questions