DE Mastery GitHub

Apache Kafka · KFK-10

Testing & observability (apps)

From your world. dbt test + CI gates, and the lag/freshness monitors you already run on a warehouse

← All exercises

Consumer-lag metrics and alerting for the copier

You own a transactional copier app. It is in app.py as copy_batch, and it already works. It reads from a source topic and copies to a destination topic exactly-once. In production, you need to watch it. Add the observability layer.

Edit app.py and implement four functions.

1. compute_lag(committed, end_offsets) -> dict[int, int] (pure)

  • committed: {partition: committed_offset}. A value of None or a negative number means the group has no committed offset for that partition (nothing processed yet).
  • end_offsets: {partition: end_offset} (the log end offset / high watermark).
  • Return {partition: lag} where lag = end_offset - committed_offset. A partition with no committed offset is fully unread (lag == end_offset). Lag is never negative.

2. total_lag(lag) -> int (pure)

Return the sum of the per-partition lag.

3. evaluate_alerts(lag_by_partition, warn_threshold, page_threshold) -> str (pure)

  • Return "page" if any partition has lag >= page_threshold.
  • Otherwise, return "warn" if any partition has lag >= warn_threshold.
  • Otherwise, return "ok".

The highest severity wins. The thresholds are inclusive.

4. measure_group_lag(bootstrap_servers, topic, group_id) -> dict[int, int]

Read the committed offsets of the group and the end offsets of the partitions from a live broker. Return the lag per partition. Do not join the group to consume. Only read the offsets (kafka-consumer-groups.sh --describe does the same thing). Reuse compute_lag.

Constraints

  • compute_lag, total_lag and evaluate_alerts must be pure (no Kafka, no I/O). The tests use plain dictionaries and an independent Python oracle to grade them.
  • The tests grade measure_group_lag against a real native broker (no Docker). The test produces N records and runs the copier over some of them. Then it asserts that the lag you report equals the true remaining backlog.

How it is graded

  • The pure functions are compared with a Python oracle over many random partition/offset combinations (property-style). This includes the edge cases of no committed offset and all caught up.
  • evaluate_alerts boundaries: just below, exactly at, and above each threshold. Also warn vs page.
  • measure_group_lag: produce 10, copy 4, and assert that the reported total lag == 6. Produce more without consuming, and assert that lag rises. Drain fully, and assert that lag == 0.

Starter code

app.py
"""Starter — KFK-10. The copier app is given and works. Your job is to add the
observability functions. Each stub below MUST be implemented (they currently
fail the tests: gate G1).
"""

from __future__ import annotations

import uuid


# ---------------------------------------------------------------------------
# PROVIDED: a transactional (exactly-once) copier. Do not change it.
# ---------------------------------------------------------------------------
def copy_batch(bootstrap_servers, src_topic, dst_topic, group_id,
               max_records=1000, poll_timeout=0.5, idle_rounds=6):
    from confluent_kafka import Consumer, Producer, TopicPartition

    consumer = Consumer({
        "bootstrap.servers": bootstrap_servers,
        "group.id": group_id,
        "enable.auto.commit": False,
        "auto.offset.reset": "earliest",
    })
    producer = Producer({
        "bootstrap.servers": bootstrap_servers,
        "enable.idempotence": True,
        "transactional.id": f"copier-{uuid.uuid4()}",
    })
    producer.init_transactions()
    consumer.subscribe([src_topic])

    copied = 0
    try:
        producer.begin_transaction()
        pending = {}  # partition -> last offset consumed
        idle = 0
        # Count idle polls only AFTER partitions are assigned, so a slow
        # rebalance never looks like an empty topic (deterministic under
        # shuffled parallel runs). No warmup poll discards data.
        while copied < max_records and idle < idle_rounds:
            msg = consumer.poll(poll_timeout)
            if msg is None:
                if consumer.assignment():
                    idle += 1
                continue
            if msg.error():
                continue
            idle = 0
            producer.produce(dst_topic, key=msg.key(), value=msg.value())
            pending[msg.partition()] = msg.offset()
            copied += 1

        if copied:
            offsets = [TopicPartition(src_topic, p, off + 1)
                       for p, off in pending.items()]
            producer.send_offsets_to_transaction(
                offsets, consumer.consumer_group_metadata())
            producer.commit_transaction()
        else:
            producer.abort_transaction()
    finally:
        consumer.close()
    return copied


# ---------------------------------------------------------------------------
# TODO: implement these.
# ---------------------------------------------------------------------------
def compute_lag(committed, end_offsets):
    """Return {partition: lag} where lag = end_offset - committed_offset.
    A partition with no committed offset (None or < 0) is fully unread.
    Lag must never be negative."""
    # TODO
    return {}


def total_lag(lag):
    """Return the sum of per-partition lag."""
    # TODO
    return 0


def evaluate_alerts(lag_by_partition, warn_threshold, page_threshold):
    """Return 'page' / 'warn' / 'ok'. Highest severity wins; inclusive."""
    # TODO
    return "ok"


def measure_group_lag(bootstrap_servers, topic, group_id, timeout=15.0):
    """Read the group's committed offsets and the partitions' end offsets from
    the broker and return lag per partition (reuse compute_lag)."""
    # TODO
    return {}

Hints

Hint 1

compute_lag: for each partition, lag = end_offset - committed_offset. A partition with no committed offset (committed is None or < 0) has lag equal to its whole end_offset (everything is unread). Never return a negative lag.

Hint 2

evaluate_alerts: look at the per-partition lags. Return 'page' if ANY partition is at/above page_threshold, else 'warn' if ANY is at/above warn_threshold, else 'ok'. The highest severity wins; thresholds are inclusive.

Hint 3

measure_group_lag: read the group's committed offsets (AdminClient.list_consumer_group_offsets with ConsumerGroupTopicPartitions, or Consumer.committed) and each partition's end offset (AdminClient.list_offsets with OffsetSpec.latest, or Consumer.get_watermark_offsets), then call compute_lag. A partition the group never committed to counts as fully unread.

Run it locally
uv run de start KFK-10/01-lag-alerting
# edit workspace/KFK-10/01-lag-alerting/
uv run de test