Alerting on Kafka Consumer Lag for Sensor Topics #

A consumer-lag alert has one job: distinguish a pipeline that is behind and recovering from one that is behind and losing. Almost every implementation gets it wrong in the same way — a fixed threshold on the topic-level sum — and the result is an alert that pages during every deploy and stays silent through the slow stall it was built for. This guide builds the alert properly: per partition, on the derivative, with a projected drain time that turns a backlog into an answerable question. It applies the signals defined in stream observability and lag monitoring.

Three Conditions That Together Mean “Losing” #

Absolutely behind. Lag exceeds what your freshness objective tolerates. Necessary, and on its own nearly useless.

Not recovering. The derivative of lag over the last several minutes is non-negative. This single condition removes almost all deploy-related noise, because a restarting consumer’s lag spikes and then falls monotonically.

Drain time exceeds the objective. Backlog divided by drain rate. If a consumer is 400 000 messages behind and clearing 200 messages per second, it needs thirty-three minutes — which is either fine or an incident depending on what the data feeds, and no threshold on the raw backlog can express that.

Combining them turns “lag is 400 000” into “this partition will not be current for 33 minutes and is not improving”, which is a statement someone can act on at 3 a.m.

Production-Ready Implementation #

Exporting the metric correctly comes first. The broker knows the high watermark and the group’s committed offset; the difference is the truth, and reading it from the admin API is far more reliable than inferring it inside the consumer.

# python 3.11 · confluent-kafka==2.4.0 · prometheus-client==0.20.0
from confluent_kafka import Consumer, TopicPartition
from confluent_kafka.admin import AdminClient
from prometheus_client import Gauge

LAG = Gauge("kafka_consumer_lag_messages", "High watermark minus committed offset",
            ["topic", "partition", "group"])
DRAIN = Gauge("kafka_consumer_drain_rate", "Messages consumed per second",
              ["topic", "partition", "group"])


def export_lag(consumer: Consumer, group: str, topic: str) -> dict[int, int]:
    """Per-partition lag from broker watermarks and the group's committed offsets."""
    md = consumer.list_topics(topic, timeout=10).topics[topic]
    parts = [TopicPartition(topic, p) for p in md.partitions]
    committed = consumer.committed(parts, timeout=10)

    lags: dict[int, int] = {}
    for tp in committed:
        _low, high = consumer.get_watermark_offsets(tp, timeout=10, cached=False)
        # An unset committed offset is -1001 (OFFSET_INVALID): the group has never
        # committed here, so the whole retained log is outstanding.
        offset = tp.offset if tp.offset >= 0 else _low
        lag = max(0, high - offset)
        lags[tp.partition] = lag
        LAG.labels(topic=topic, partition=str(tp.partition), group=group).set(lag)
    return lags

Handling OFFSET_INVALID explicitly is not a nicety. A newly created group reports offset -1001, and subtracting it from the high watermark produces a lag larger than the topic, which fires every alert simultaneously the first time a consumer group is deployed.

The drain rate is measured, not assumed:

import time
from collections import deque


class DrainRate:
    """Sliding-window consumption rate, per partition."""

    def __init__(self, window_s: float = 120.0) -> None:
        self.window_s = window_s
        self._samples: dict[int, deque[tuple[float, int]]] = {}

    def observe(self, partition: int, committed_offset: int) -> float | None:
        now = time.monotonic()
        buf = self._samples.setdefault(partition, deque())
        buf.append((now, committed_offset))
        while buf and now - buf[0][0] > self.window_s:
            buf.popleft()
        if len(buf) < 2:
            return None
        (t0, o0), (t1, o1) = buf[0], buf[-1]
        elapsed = t1 - t0
        return (o1 - o0) / elapsed if elapsed > 0 else None


def projected_drain_seconds(lag: int, rate: float | None) -> float:
    """How long until this partition is current, at the observed rate."""
    if not rate or rate <= 0:
        return float("inf")          # not draining at all
    return lag / rate

float("inf") for a stalled consumer is deliberate: it makes “not draining” the maximum value in any comparison, so a partition that has stopped entirely always sorts above one that is merely slow.

The alerting rules then express the three conditions:

# prometheus rules
groups:
  - name: sensor-stream-lag
    rules:
      - alert: SensorStreamNotRecovering
        expr: |
          max by (topic, partition) (kafka_consumer_lag_messages) > 50000
          and
          deriv(max by (topic, partition) (kafka_consumer_lag_messages)[10m:]) >= 0
        for: 10m
        labels: {severity: page}
        annotations:
          summary: " partition  is behind and not recovering"

      - alert: SensorStreamPartitionStalled
        expr: |
          rate(kafka_consumer_drain_rate[5m]) == 0
          and max by (topic, partition) (kafka_consumer_lag_messages) > 1000
        for: 5m
        labels: {severity: page}

      - alert: SensorStreamDrainSlow
        expr: |
          max by (topic) (kafka_consumer_lag_messages)
            / clamp_min(max by (topic) (kafka_consumer_drain_rate), 1) > 1800
        for: 15m
        labels: {severity: ticket}

The third rule is the one that catches the slow stall that started this page: a pipeline whose drain time has crept past thirty minutes without ever crossing a dramatic message-count threshold.

Parameter Tuning Guide #

The for: column is the noise control. A rebalance stalls a group for tens of seconds; a deployment for a minute or two. Anything below a five-minute for: will page on both, and the rebalancing guide explains why those events are routine rather than exceptional.

Verification and Testing #

Test the exporter against the two edge cases that produce spurious pages, then test the rules with a replayed series.

# python 3.11 · pytest==8.2.0
def test_uncommitted_group_does_not_report_the_whole_topic_as_lag(consumer, topic):
    """A brand-new consumer group must not report lag equal to the retained log."""
    lags = export_lag(consumer, group="brand-new-group", topic=topic)
    assert all(lag <= 1 for lag in lags.values())


def test_stalled_partition_reports_infinite_drain_time():
    rate = DrainRate(window_s=60)
    rate.observe(0, 1_000)
    rate.observe(0, 1_000)                       # committed offset has not moved
    assert projected_drain_seconds(500_000, rate.observe(0, 1_000)) == float("inf")


def test_recovering_backlog_is_not_alertable():
    """Falling lag must fail the not-recovering condition even while above threshold."""
    series = [900_000, 700_000, 520_000, 340_000, 180_000]
    assert all(b < a for a, b in zip(series, series[1:]))    # strictly decreasing
    # deriv(...) < 0 → SensorStreamNotRecovering does not fire despite lag > 50 000

Beyond unit tests, run a game day: stop one consumer replica and confirm exactly one alert fires, naming the right partition, within the configured window. Then deploy a no-op change and confirm nothing fires. An alert that survives both is worth keeping.

Gotchas #

Summing lag across partitions. The topic-level sum hides a single stalled partition inside twenty-three healthy ones and is the single most common reason a real stall goes unnoticed. Use max by (partition).

Alerting on lag for a group with no consumers. A retired consumer group keeps its committed offsets, so its lag grows forever and pages nightly until someone deletes the group. Alert only on groups with active members.

Measuring drain rate from the consumer’s own counters. A stalled consumer stops updating its own metrics, so its reported rate freezes at the last healthy value rather than falling to zero. Derive the rate from committed offsets observed externally, as above.

cached=True on watermark lookups. It is the default in some client wrappers and returns a stale high watermark, which makes lag appear to shrink during the exact incident you are watching. Pass cached=False when the number is going to drive an alert.

FAQ #

Related #

• Stream Observability and Lag Monitoring — the observability stage this alert belongs to
• Measuring End-to-End Latency in a Sensor Pipeline — the per-stage breakdown you consult once the alert fires
• Managing Consumer Group Rebalancing for Sensor Topics — the most common benign cause of a lag spike