Monitoring Kafka Consumer Lag for DSR SLAs

When Data Subject Request (DSR) discovery runs through a Kafka topic, the single number that predicts whether you will meet a statutory deadline is not CPU or memory — it is consumer-group lag, the count of messages produced but not yet processed. Within the broader Async Polling & Queue Management approach, and the wider Cross-System Data Discovery & Sync stage it belongs to, lag is the early-warning instrument: a backlog that is merely large is tolerable, but a backlog that will take longer to drain than the time left on the clock is a looming breach of the “without undue delay and in any event within one month” mandate of GDPR Article 12(3). This page shows privacy and platform engineers how to measure per-partition lag with kafka-python, convert it into an estimated time-to-drain, and alert only when that drain estimate actually threatens the remaining SLA budget. It complements Implementing Celery for async polling, which distributes the work, and Configuring exponential backoff within SLA budgets, which governs how retries spend that same budget.

Prerequisites #

• Python 3.11+ — for zoneinfo, PEP 604 unions (float | None), and datetime.UTC.
• kafka-python 2.0+ (pip install kafka-python) for KafkaConsumer and KafkaAdminClient. The consumer exposes end_offsets(); the admin client exposes list_consumer_group_offsets().
• Pydantic v2 (pydantic>=2.6) for the typed LagReport emitted to your alerting layer.
• A Kafka cluster reachable over TLS, the consumer-group id under observation, and read access to __consumer_offsets (granted implicitly to the admin client with Describe on the group).
• A source of truth for each request’s statutory deadline. This monitor treats the deadline as an input; computing it correctly across jurisdictions is covered upstream in the intake layer.

Measurement must be read-only. The monitor never joins the group being observed and never commits offsets — doing so would move the very cursor it is trying to measure and corrupt the backlog figure.

Step-by-step implementation #

Step 1 — Compute per-partition lag #

Lag is the log-end offset (the next offset a producer will write) minus the committed offset (the last position the group acknowledged) for every partition the group owns. Summing across partitions hides a stuck partition behind healthy ones, so keep the figure per-partition and aggregate only for display.

from kafka import KafkaAdminClient, KafkaConsumer
from kafka.structs import TopicPartition


def read_lag(bootstrap: str, group_id: str, topic: str) -> dict[TopicPartition, int]:
    """Return per-partition lag (log-end minus committed) without joining the group.

    The admin client reads committed offsets from __consumer_offsets; a throwaway
    consumer (not a member of ``group_id``) reads the log-end high-water marks.
    """
    admin = KafkaAdminClient(bootstrap_servers=bootstrap)
    consumer = KafkaConsumer(bootstrap_servers=bootstrap, enable_auto_commit=False)
    try:
        committed = admin.list_consumer_group_offsets(group_id)  # {tp: OffsetAndMetadata}
        partitions = [tp for tp in committed if tp.topic == topic]
        end_offsets = consumer.end_offsets(partitions)           # {tp: log-end offset}

        lag: dict[TopicPartition, int] = {}
        for tp in partitions:
            committed_offset = committed[tp].offset
            # A group that never committed reports offset -1; treat the whole
            # backlog as unread rather than a nonsensical negative lag.
            if committed_offset < 0:
                lag[tp] = end_offsets[tp]
            else:
                lag[tp] = max(0, end_offsets[tp] - committed_offset)
        return lag
    finally:
        consumer.close()
        admin.close()

Step 2 — Estimate throughput #

An ETA needs a rate. Sample the committed offset twice, divide the delta by the elapsed wall-clock seconds, and smooth the result with an exponentially weighted moving average so a single quiet interval does not make the ETA jump to infinity. Smoothing is what makes the alert stable enough to page on.

from dataclasses import dataclass, field


@dataclass
class ThroughputTracker:
    """Exponentially weighted moving average of processed messages per second."""

    alpha: float = 0.3            # smoothing factor; higher reacts faster
    rate: float | None = field(default=None)
    _last_total: int | None = field(default=None)
    _last_ts: float | None = field(default=None)

    def update(self, committed_total: int, now: float) -> float | None:
        """Fold one (total committed, timestamp) sample into the EWMA rate."""
        if self._last_total is not None and self._last_ts is not None:
            elapsed = now - self._last_ts
            processed = committed_total - self._last_total
            if elapsed > 0 and processed >= 0:
                sample = processed / elapsed
                self.rate = sample if self.rate is None else (
                    self.alpha * sample + (1 - self.alpha) * self.rate
                )
        self._last_total, self._last_ts = committed_total, now
        return self.rate

Step 3 — Convert lag to a drain ETA #

The drain ETA is total lag divided by the smoothed throughput. Guard the zero-throughput case explicitly: a backlog with no forward progress has no finite ETA, and returning None here lets the alert logic treat “stuck” distinctly from “slow”.

def drain_eta_seconds(total_lag: int, rate: float | None) -> float | None:
    """Seconds to clear the backlog at the current rate, or None if stalled."""
    if total_lag == 0:
        return 0.0
    if rate is None or rate <= 0:
        return None                # stuck: no forward progress to extrapolate
    return total_lag / rate

Step 4 — Raise an SLA-aware alert threshold #

Absolute-lag thresholds are the classic mistake: a 500k backlog draining at 50k/s clears in ten seconds, while a 5k backlog draining at 1/s never clears in time. Alert on the relationship between the ETA and the remaining budget — the statutory deadline minus now — with a safety margin so the page fires while there is still time to intervene, not at the moment of breach. Under GDPR Article 12(3) the controller must act “without undue delay,” so the margin should leave room for manual escalation.

from datetime import datetime


def sla_at_risk(
    eta_seconds: float | None,
    deadline: datetime,
    now: datetime,
    safety: float = 0.75,
) -> bool:
    """True when the projected drain time threatens the remaining SLA window.

    ``safety`` reserves headroom: at 0.75 the alert fires once the ETA exceeds
    75% of the time left, leaving a quarter of the budget for human response.
    A stalled backlog (eta_seconds is None) is always at risk.
    """
    remaining = (deadline - now).total_seconds()
    if remaining <= 0:
        return True                # already past the statutory deadline
    if eta_seconds is None:
        return True                # stuck consumer: cannot promise completion
    return eta_seconds > remaining * safety

Step 5 — Emit a typed LagReport #

Alerting layers, dashboards, and the audit trail all consume the same structured record. A Pydantic v2 model gives it a validated shape and keeps subject identifiers out — only offsets, rates, and derived timings belong in an operational metric.

from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator


class LagReport(BaseModel):
    """A single SLA-aware snapshot of consumer-group health."""

    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    group_id: str
    topic: str
    total_lag: int = Field(ge=0)
    throughput_msgs_s: float | None = Field(default=None, ge=0)
    drain_eta_seconds: float | None = Field(default=None, ge=0)
    remaining_budget_seconds: float
    at_risk: bool
    observed_at: datetime

    @field_validator("group_id", "topic")
    @classmethod
    def not_blank(cls, v: str) -> str:
        """Reject empty identifiers so a misconfigured probe never masquerades as healthy."""
        if not v:
            raise ValueError("group_id and topic must be non-empty")
        return v


def build_report(
    group_id: str,
    topic: str,
    lag_by_partition: dict,
    rate: float | None,
    deadline: datetime,
    now: datetime,
) -> LagReport:
    """Assemble the report the alerting layer and audit trail both consume."""
    total = sum(lag_by_partition.values())
    eta = drain_eta_seconds(total, rate)
    return LagReport(
        group_id=group_id,
        topic=topic,
        total_lag=total,
        throughput_msgs_s=rate,
        drain_eta_seconds=eta,
        remaining_budget_seconds=(deadline - now).total_seconds(),
        at_risk=sla_at_risk(eta, deadline, now),
        observed_at=now,
    )

Configuration reference #

Verification #

Assert the two decisions that matter: a large-but-fast backlog stays quiet, and a small-but-stalled backlog pages.

from datetime import datetime, timedelta, UTC


def test_fast_backlog_is_not_at_risk() -> None:
    """A 500k backlog draining at 50k/s (10s ETA) with a full day left is fine."""
    now = datetime.now(UTC)
    eta = drain_eta_seconds(total_lag=500_000, rate=50_000)
    assert eta is not None and eta < 11
    assert sla_at_risk(eta, deadline=now + timedelta(days=1), now=now) is False


def test_stalled_consumer_is_always_at_risk() -> None:
    """Zero throughput yields no finite ETA and must always page."""
    now = datetime.now(UTC)
    assert drain_eta_seconds(total_lag=5_000, rate=0.0) is None
    assert sla_at_risk(None, deadline=now + timedelta(hours=6), now=now) is True

Expected: a healthy run emits a LagReport with at_risk=False; when throughput collapses, the next report flips to at_risk=True while drain_eta_seconds is None.

Troubleshooting #

Lag spikes then collapses during a rebalance When a consumer group rebalances, partitions are momentarily unassigned and committed offsets can read stale, producing a lag figure that jumps and then snaps back. Fix: ignore reports captured within a cooldown of the last GroupCoordinator rebalance, or require two consecutive at-risk samples before paging so a rebalance transient cannot fire a false alarm.

Consumer is stuck but lag looks flat, not growing If producers have also paused, a wedged consumer shows constant lag rather than rising lag, and an absolute-threshold alert stays silent. Fix: this is exactly why Step 3 returns None for zero throughput and Step 4 treats None as always-at-risk — flat lag with zero rate is a stall, not health.

Negative or absurd lag values A group that has never committed reports offset -1, and a naive subtraction yields a nonsensical negative or gigantic lag. Fix: the committed_offset < 0 branch in Step 1 treats an uncommitted partition as fully unread instead.

Drain ETA disagrees with the deadline by hours Clock skew between the monitor host and the deadline source makes remaining_budget_seconds wrong. Fix: compute both now and deadline as timezone-aware UTC (datetime.now(UTC)), never naive local time, and keep the monitor host on NTP so the two frames agree.

Alert fires on a topic that is simply idle A quiet topic has near-zero throughput, so the EWMA can read stuck even though there is no backlog. Fix: short-circuit when total_lag == 0 (Step 3 returns 0.0), so an empty backlog is never at risk regardless of throughput.

Frequently Asked Questions #

Related #

• Async Polling & Queue Management — the parent stage: bounded ingestion, priority routing, and failure categorization this lag monitor watches over.
• Implementing Celery for async polling — the distributed worker fleet whose queue depth this monitor translates into an SLA risk.
• Designing dead-letter queue retry for DSR workers — where messages that cannot drain are routed so lag does not silently swallow a request.
• Configuring exponential backoff within SLA budgets — how retry timing spends the same budget this monitor guards.
• Cross-System Data Discovery & Sync — the discovery stage this queue telemetry belongs to.