Four Million Echoes

A medium AI Coding interview practice problem on DataDriven.

Domain
AI Coding
Difficulty
medium
Seniority
L5
Asked in
Confluent

Problem

Our orders enrichment consumer reprocesses ~4M messages every time we deploy. The team added enable_auto_commit=False and explicit commit() calls weeks ago, so this should have stopped. Run tests/test_offsets_persist.py to reproduce. The reprocess window correlates with rebalance events. Find why the explicit commits aren't sticking and fix it. Bonus credit: explain what 'cooperative-sticky' would have changed.

It happened. It happened again.

Worked solution and explanation

This is a commit-before-revoke bug wearing a 'why do we replay 4M rows on deploy' costume. The real skill: knowing that enable_auto_commit=False only moves the durability decision into your hands, it doesn't make offsets stick. Everyone gets the batch loop right. The thing that separates seniors is realizing that an eager rebalance rips your partitions away mid-batch, and the broker never calls back into run() to give you a chance to flush. Miss that and every deploy replays the last uncommitted batch on every rebalanced partition, which is exactly the ~4M-echo window in the ticket.

Why the explicit commits don't stick

Trace the timeline. run() polls 550 messages: one full batch of 500 is enriched, written and committed, then 50 more sit in batch (and in the wrapper's _held) uncommitted. A deploy triggers a rebalance. With partition.assignment.strategy=range (eager), the broker revokes ALL partitions stop-the-world and reassigns them, then the new owner resumes from the last committed offset. Those 50 in-flight messages were never committed, so they get redelivered. Multiply by every partition, every rebalance, every deploy: that's your replay storm. The commit() calls you added are correct, there just aren't enough of them, one is owed at revoke time.

The starter (broken)
def run(consumer, max_messages):
    processed = 0
    batch = []
    for msg in consumer:
        batch.append(msg)
        if len(batch) >= BATCH_SIZE:
            enriched = [enrich(m) for m in batch]
            write_enriched(enriched)
            consumer.commit()
            processed += len(batch)
            batch = []
        if processed >= max_messages:
            break
    # ... trailing flush, but no on_partitions_revoked hook

Correct batching, but nothing runs at revoke: the 50 in-flight messages die here.

The fix: flush during the revoke callback

Step 1: Pull the flush into one closure

The commit logic now runs from two places, the batch boundary and the revoke callback, so give it one name. A flush() closure over batch and processed keeps the ordering in a single spot and avoids duplicating the enrich/write/commit sequence.

Step 2: Register on_partitions_revoked before you poll

The KafkaConsumer wrapper already exposes the hook; run() just has to set consumer.on_partitions_revoked = flush. Now trigger_rebalance() calls it while the batch is still in flight, giving you the one commit you owed the broker before reassignment.

Step 3: Write to the sink BEFORE you commit, and no-op on empty

Order matters: write_enriched(...) then consumer.commit(). Commit first and a sink failure loses rows you've already marked consumed. And guard on empty, if not batch: return, so a rebalance that fires on a clean batch boundary doesn't re-commit stale offsets (the test_no_double_commit_on_clean_batch edge).

consumer/run.py (fixed)
from consumer.client import KafkaConsumer
from consumer.warehouse import write_enriched
from consumer.enrich import enrich

BATCH_SIZE = 500


def run(consumer: KafkaConsumer, max_messages: int) -> int:
    processed = 0
    batch = []

    def flush():
        nonlocal batch, processed
        if not batch:
            return
        write_enriched([enrich(m) for m in batch])
        consumer.commit()
        processed += len(batch)
        batch = []

    # Eager rebalance revokes partitions stop-the-world. Flush the
    # in-flight batch here, before the broker reassigns them.
    consumer.on_partitions_revoked = flush

    for msg in consumer:
        batch.append(msg)
        if len(batch) >= BATCH_SIZE:
            flush()
        if processed >= max_messages:
            break
    flush()
    return processed

One flush() closure, wired to on_partitions_revoked, write before commit, empty-safe.

eager (range)
Every revoke is stop-the-world: ALL partitions are given up and reassigned in one shot. Any consumer holding an uncommitted batch on ANY partition must flush in the revoke callback or lose it.
cooperative-sticky
Only the partitions actually moving are revoked; the rest keep flowing, so a deploy stops far less work. But it does NOT solve commit-before-revoke: the partitions that DO move still need a flush in on_partitions_revoked. It shrinks the blast radius, it doesn't remove the callback.

Common follow-up questions