Build your own mini stream processor · Stage 8 of 8

Transactional (2PC) sink — end-to-end exactly-once

Exactly-once state (stage 7) is not enough for exactly-once output: the sink must not publish records that belong to a checkpoint that never completed. Flink's TwoPhaseCommitSinkFunction ties sink transactions to checkpoints — pre-commit on the barrier, commit on the checkpoint-complete notification, and abort anything not yet committed on recovery.

This stage is optional for the learner, but it is the capstone of the EOS story and is fully graded here.

New API this stage

TransactionalSink()      # write / pre_commit(cid) / commit(cid) / abort(cid) / abort_open()
run_exactly_once_sink(records, barrier_every, crash_after, commit_lag=0) -> list

This stage's task

TransactionalSink: write buffers into the open transaction; pre_commit(cid) moves it to prepared[cid] (durable, invisible); commit(cid) appends prepared[cid] to the visible committed list exactly once; abort(cid) and abort_open() discard.
run_exactly_once_sink: on each barrier, pre-commit and queue a pending commit; a commit notification arrives commit_lag barriers later. On a clean finish, do a final pre-commit of the trailing open transaction and flush all pending commits.
On crash: abort the open transaction and every prepared-but-unnotified transaction, then replay the source from the last completed checkpoint's offset and commit that tail.

The EOS invariant (how it is graded)

For every crash point and commit lag 0/1, the sorted committed output must equal the batch oracle — every input appears exactly once, with no duplicates (uncommitted staging is aborted and replayed) and no loss.

Constraints

Deterministic crash schedule, idempotent commit, no sleeps. Earlier stages still pass. Suite < 90 s.

Run it

de start FLK-BYO/08
de test

Hints

Hint 1

TransactionalSink models Flink's TwoPhaseCommitSinkFunction. write() buffers into the OPEN transaction. pre_commit(cid) flushes the open txn into prepared[cid] (durable, invisible) and clears the open txn. commit(cid) appends prepared[cid] to the visible `committed` list EXACTLY ONCE (ignore a duplicate/absent cid). abort(cid) drops prepared[cid]; abort_open() drops the open txn.

Hint 2

run_exactly_once_sink injects barriers; on each Barrier pre_commit the open txn (record the source offset) and queue a pending commit; a commit notification lands commit_lag barriers later (pop the oldest pending and commit it). When the job finishes cleanly, do a FINAL pre_commit of the trailing open txn and flush all pending commits.

Hint 3

On crash: abort the open txn AND any prepared txn whose checkpoint was never notified (aborted = not visible). Then replay the source from the last COMPLETED checkpoint's offset into a fresh sink, pre_commit + commit that tail, and the visible output = committed-before-crash + replayed tail. For every crash point and lag the sorted committed output must equal the batch oracle (each input exactly once).

Run it locally

uv run de start FLK-BYO/08-transactional-sink
uv run de test