DE Mastery GitHub

Apache Kafka · KFK-04

Consumers & groups — manual commits, at-least-once, no loss across crashes

From your world. SOX completeness: a crash must never drop an unprocessed event

← All exercises

At-least-once consumer with manual commits — no loss across a crash

Build a consumer that never loses a record, even if the process crashes in the middle of a stream. The guarantee comes only from committing the offset after the work, not before.

Implement in solution.py:

def process_batch(bootstrap_servers, topic, group_id, max_records, sink) -> int:
    ...
  • Create a confluent_kafka.Consumer with the given group.id, enable.auto.commit=False, and auto.offset.reset="earliest". Subscribe to topic.
  • For each message, do these steps in this exact order:
    1. Call sink(key, value). key and value are the decoded UTF-8 strings. This call is "the work".
    2. Only if sink returned normally, commit the offset of that message (consumer.commit(message=msg, asynchronous=False)).
  • Stop when you have processed and committed max_records messages. Also stop when a short idle period passes and no new message arrives. Return the number of messages that you processed. Close the consumer.
  • If sink raises an exception, let the exception propagate and do not commit that record. This models a crash. The record stays uncommitted. Kafka redelivers it to the next run of the group.

The committed offset moves only past records for which sink succeeded. So after a crash and a resume (same group_id), you reprocess from the last commit. You never skip a record (at-least-once). A record may be delivered more than once.

The tests produce a known stream. They run process_batch once with a sink that raises partway through (the crash). Then they run it again with a normal sink. They assert two things. First, the code handled every produced record at least once (no loss). Second, the crash caused a redelivery (at-least-once).

Starter code

solution.py
"""Starter — not implemented yet (must FAIL the tests: gate G1). See README.md."""


def process_batch(bootstrap_servers, topic, group_id, max_records, sink):
    # TODO: consume up to max_records, call sink(key, value) FIRST, then commit
    #       that offset only on success. Return the number processed.
    return 0

Hints

Hint 1

Create a Consumer with the given group.id, enable.auto.commit=False and auto.offset.reset='earliest'. Subscribe to the topic.

Hint 2

For each polled message: FIRST call sink(key, value) to do the work, and ONLY if it returns normally commit that message's offset (consumer.commit(message=msg, asynchronous=False)). This ordering is what makes it at-least-once: a crash in sink() leaves the offset uncommitted, so the record is redelivered.

Hint 3

Stop after you have successfully processed-and-committed max_records messages (or a short idle timeout elapses). Return the number you processed. If sink() raises, let it propagate WITHOUT committing that record (the crash).

Run it locally
uv run de start KFK-04/01-at-least-once
# edit workspace/KFK-04/01-at-least-once/
uv run de test