# How to Monitor and Alert on Kafka Consumer Lag Metrics in the Data Engineering Zoomcamp

> Learn to monitor and alert on Kafka consumer lag using KafkaConsumer API, Prometheus for Spark, or Confluent Cloud integrations. Essential for data engineers.

- Repository: [DataTalksClub/data-engineering-zoomcamp](https://github.com/DataTalksClub/data-engineering-zoomcamp)
- Tags: how-to-guide
- Published: 2026-05-31

---

**You can monitor and alert on Kafka consumer lag metrics by polling offsets via the Python `KafkaConsumer` API, scraping JMX beans with Prometheus for Spark workloads, or leveraging Confluent Cloud’s native Grafana integration depending on whether you run self-hosted brokers or managed infrastructure.**

The DataTalksClub/data-engineering-zoomcamp repository demonstrates real-time streaming pipelines that ingest data using either **Spark Structured Streaming** ([[`streaming_confluent.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/streaming_confluent.py)](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2023/week_6_stream_processing/streaming_confluent.py)) or the **`kafka-python`** library ([[`consumer.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/consumer.py)](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/07-streaming/workshop/src/consumers/consumer.py)). To prevent downstream staleness and broken SLAs, you must monitor and alert on Kafka consumer lag metrics—the difference between a partition’s latest offset (high-water mark) and the last committed offset of your consumer group.

## Understanding Consumer Lag in the Zoomcamp Architecture

Consumer lag indicates how far behind your application is from the tip of the stream. In the Zoomcamp codebase, Kafka connections are centralized in [[`settings.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/settings.py)](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/cohorts/2023/week_6_stream_processing/settings.py) via the `CONFLUENT_CLOUD_CONFIG` dictionary, which supplies bootstrap servers and SASL credentials to multiple components:

- **Spark Streaming**: Configured at lines 13–14 of [`streaming_confluent.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/streaming_confluent.py) with `.option("kafka.bootstrap.servers", CONFLUENT_CLOUD_CONFIG['bootstrap.servers'])`
- **Python Consumer**: Configured at lines 13–16 of [`07-streaming/workshop/src/consumers/consumer.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/07-streaming/workshop/src/consumers/consumer.py) using `KafkaConsumer(..., bootstrap_servers=[server])`
- **Confluent Producer**: Initialized at line 16 of [`producer_confluent.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/producer_confluent.py) via `Producer(**probs)`

Each of these components can expose lag metrics, but the extraction method varies by runtime and library.

## Four Methods to Monitor and Alert on Kafka Consumer Lag Metrics

### Confluent Cloud Native Metrics and Grafana

If you deploy against **Confluent Cloud** as configured in [`settings.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/settings.py), enable the built-in `ccloud_consumer_lag` Prometheus metric from the Confluent Cloud UI. Connect Grafana to the Confluent Cloud Prometheus endpoint and create alert thresholds directly in the UI. This approach requires no additional code changes to the Zoomcamp scripts.

### JMX Exporter and Prometheus for Self-Hosted Clusters

For self-hosted brokers or Java-based consumers like Spark, deploy the **JMX Exporter** to scrape the `kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,topic=*,partition=*` MBean. This exposes the `records-lag-max` gauge. Prometheus scrapes these values from `http://<broker-host>:9999/metrics`, and Alertmanager fires alerts when thresholds are breached.

### Python Admin API Polling for Custom Scripts

When using the `kafka-python` library shown in [`consumer.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/consumer.py), implement a lightweight health-check script that compares **end offsets** (high-water mark) with **committed offsets**. This method works for any Python runtime and can push metrics to Prometheus, Datadog, or logs without requiring a JVM. The script reuses the same `bootstrap_servers` and `group_id` configuration found in [`consumer.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/consumer.py) (line 17 uses `rides-console`).

### Confluent-Kafka Library Metrics Method

If you use the `confluent_kafka` Python library (demonstrated in [`producer_confluent.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/producer_confluent.py)), instantiate a `Consumer` and call its `.metrics()` method. This returns a dictionary containing `consumer_lag` per topic-partition, which you can serialize and expose via an HTTP endpoint for scraping.

## Building a Custom Python Lag Monitor

The following script, compatible with the `kafka-python` consumer in [`07-streaming/workshop/src/consumers/consumer.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/07-streaming/workshop/src/consumers/consumer.py), polls lag for every partition and prints the delta. You can extend it to push metrics to Prometheus by uncommenting the optional instrumentation section.

```python

# monitor_consumer_lag.py

import os
from kafka import KafkaConsumer, TopicPartition

# Settings – reuse the same server and topic as consumer.py

BOOTSTRAP_SERVER = os.getenv("KAFKA_BOOTSTRAP", "localhost:9092")
TOPIC = os.getenv("KAFKA_TOPIC", "rides")
GROUP_ID = os.getenv("KAFKA_GROUP", "rides-console")

consumer = KafkaConsumer(
    bootstrap_servers=[BOOTSTRAP_SERVER],
    group_id=GROUP_ID,
    enable_auto_commit=False,   # we only need metadata, not actual consumption

)

# Build a list of TopicPartition objects for the target topic

partitions = consumer.partitions_for_topic(TOPIC)
if not partitions:
    raise RuntimeError(f"No partitions found for topic {TOPIC}")

tps = [TopicPartition(TOPIC, p) for p in partitions]

# Get latest offsets (high‑water mark) from the broker

latest_offsets = consumer.end_offsets(tps)

# Get the last committed offsets for the consumer group

committed_offsets = {tp: consumer.committed(tp) for tp in tps}

print(f"Consumer lag for group '{GROUP_ID}' on topic '{TOPIC}':")
total_lag = 0
for tp in tps:
    latest = latest_offsets[tp]
    committed = committed_offsets[tp].offset if committed_offsets[tp] else 0
    lag = latest - committed
    total_lag += lag
    print(f"  Partition {tp.partition}: latest={latest}, committed={committed}, lag={lag}")

print(f"Total lag across partitions: {total_lag}")

# Optionally, push to Prometheus:

# from prometheus_client import Gauge, start_http_server

# LAG_GAUGE = Gauge('kafka_consumer_lag', 'Kafka consumer lag', ['topic', 'partition'])

# for tp in tps:

#     LAG_GAUGE.labels(topic=TOPIC, partition=tp.partition).set(latest_offsets[tp] - (committed_offsets[tp].offset if committed_offsets[tp] else 0))

# start_http_server(8000)

# while True: time.sleep(5)

```

This script aligns with the existing Zoomcamp configuration by importing the same bootstrap server and group ID defined in [`consumer.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/consumer.py), allowing you to run it alongside your streaming workload without configuration drift.

## Configuring Prometheus Alert Rules

When using the JMX Exporter for Java-based consumers, add the following rule to your [`jmx_exporter.yml`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/jmx_exporter.yml) to capture per-partition lag:

```yaml
rules:
  - pattern: kafka.consumer<type=consumer-fetch-manager-metrics, client-id=.*, topic=([^,]*), partition=([0-9]+)><>records-lag-max
    name: kafka_consumer_lag
    labels:
      topic: "$1"
      partition: "$2"
    type: GAUGE

```

Then configure Alertmanager to trigger on aggregate lag exceeding your SLA threshold:

```yaml
- alert: KafkaConsumerLagHigh
  expr: sum(kafka_consumer_lag) by (topic) > 10000
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "Consumer lag for {{ $labels.topic }} is high"
    description: "Total lag across partitions is {{ $value }} messages."

```

## Summary

- **Confluent Cloud deployments** should use the native `ccloud_consumer_lag` metric and Grafana integration for zero-code monitoring.
- **Self-hosted Kafka or Spark Streaming** requires the JMX Exporter to surface `records-lag-max` beans to Prometheus, with Alertmanager handling threshold breaches.
- **Python-native workloads** using `kafka-python` can leverage the `end_offsets()` and `committed()` API methods to calculate lag programmatically and push to any metrics backend.
- The Zoomcamp repository provides reusable connection settings in [`settings.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/settings.py) and [`consumer.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/consumer.py) that allow monitoring scripts to inherit bootstrap server and credential configuration without duplication.

## Frequently Asked Questions

### What is Kafka consumer lag?

Kafka consumer lag is the number of messages a consumer group has yet to process, calculated as the difference between the partition’s latest offset (high-water mark) and the consumer’s last committed offset. Growing lag indicates that your processing rate is slower than the production rate, risking stale data and SLA violations.

### How do I check consumer lag programmatically in Python?

Use the `KafkaConsumer` class from the `kafka-python` library to call `end_offsets(partitions)` for the latest broker offsets and `committed(partition)` for the consumer group’s position. Subtract the committed offset from the end offset for each `TopicPartition` to derive the lag, as demonstrated in the [`monitor_consumer_lag.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/monitor_consumer_lag.py) script.

### Can I monitor lag for Spark Structured Streaming jobs?

Yes. Because Spark Structured Streaming uses the Kafka consumer embedded in the Spark runtime, expose the JVM’s JMX beans via the JMX Exporter. Scrape the `kafka.consumer:type=consumer-fetch-manager-metrics` MBean for `records-lag-max`, which Spark exposes automatically when reading from Kafka.

### What is a dangerous threshold for consumer lag?

A dangerous threshold depends on your SLA, but common production rules alert when lag exceeds 10,000 messages for five consecutive minutes or when the lag growth rate indicates the consumer will never catch up. For real-time pipelines in the Zoomcamp workshops, even a lag of 1,000 messages may signal backpressure requiring investigation.