DE Mastery GitHub

Apache Kafka · KFK-05

Transactions & exactly-once — read-process-write, read_committed, fencing

From your world. SOX accuracy & completeness: no duplicated or dropped ledger entries

← All exercises

Exactly-once transactional copier with ledger invariants

Copy records from src_topic to dst_topic with exactly-once semantics. Every source record appears in the destination once and only once. An aborted copy leaves nothing visible.

Implement in solution.py:

def copy_exactly_once(bootstrap_servers, src_topic, dst_topic, group_id,
                      transactional_id, max_records) -> int:
    ...

def copy_then_abort(bootstrap_servers, src_topic, dst_topic, group_id,
                    transactional_id, max_records) -> int:
    ...

copy_exactly_once(...)

This is a transactional read-process-write loop:

  1. Build a transactional Producer with transactional.id = transactional_id and enable.idempotence = True. Call init_transactions() once.
  2. Build a Consumer on group_id with enable.auto.commit=False, auto.offset.reset="earliest", and isolation.level="read_committed". Subscribe to src_topic.
  3. Call begin_transaction(). Poll up to max_records source records. produce each record (same key and value) to dst_topic. Then call send_offsets_to_transaction(consumer.position(consumer.assignment()), consumer.consumer_group_metadata()) and commit_transaction().
  4. Return the number of records that you copied. Close the consumer.

copy_then_abort(...)

This function is identical, but it calls abort_transaction() and not commit_transaction(). Return the number of records that you produced and then aborted. A read_committed consumer of dst_topic must then see none of these records (all-or-nothing).

Keep transactional_id exactly as given (stable). Never randomize it, so that zombie fencing works.

The tests produce a known set to src_topic and run copy_exactly_once. Then they read dst_topic with a read_committed consumer and assert the ledger invariants. These are: no loss (every source value is present) and no duplicate (each value appears exactly once). A separate test runs copy_then_abort and asserts that the destination has no committed records from the aborted transaction.

Starter code

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


def copy_exactly_once(bootstrap_servers, src_topic, dst_topic, group_id,
                      transactional_id, max_records):
    # TODO: transactional read-process-write; commit output + input offsets atomically.
    return 0


def copy_then_abort(bootstrap_servers, src_topic, dst_topic, group_id,
                    transactional_id, max_records):
    # TODO: same as copy_exactly_once but abort_transaction() at the end.
    return 0

Hints

Hint 1

Build a transactional Producer: {'bootstrap.servers':..., 'transactional.id': transactional_id, 'enable.idempotence': True}. Call init_transactions() once. Build a Consumer with the group.id, enable.auto.commit=False, auto.offset.reset='earliest', isolation.level='read_committed'; subscribe to src_topic.

Hint 2

copy loop: begin_transaction(); poll and produce each source record to dst_topic (same key/value); when you have the batch, call producer.send_offsets_to_transaction(consumer.position(consumer.assignment()), consumer.consumer_group_metadata()) then commit_transaction(). This makes the output writes AND the input offset commit atomic.

Hint 3

copy_then_abort does the same begin+produce but calls abort_transaction() instead of commit — a read_committed consumer must then see NONE of those records. Keep transactional_id STABLE (as given), never random, so zombie fencing works.

Run it locally
uv run de start KFK-05/01-exactly-once-copier
# edit workspace/KFK-05/01-exactly-once-copier/
uv run de test