Kafka Debug: Four Million Echoes

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.

consumer/run.py

# consumer/run.py
# Streaming consumer that enriches orders with user_country and writes
# to the warehouse. Auto-commit was disabled weeks ago (see git blame on
# this file) but offsets still don't survive a rebalance.

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 = []
    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
    if batch:
        enriched = [enrich(m) for m in batch]
        write_enriched(enriched)
        consumer.commit()
        processed += len(batch)
    return processed

# consumer/run.py
# Streaming consumer that enriches orders with user_country and writes
# to the warehouse. Auto-commit was disabled weeks ago (see git blame on
# this file) but offsets still don't survive a rebalance.

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 = []
    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
    if batch:
        enriched = [enrich(m) for m in batch]
        write_enriched(enriched)
        consumer.commit()
        processed += len(batch)
    return processed

Active Now|Sr. Data Engineer (L5)|||6.3k Attempts|2.9k Solves|