Two-Phase Commit Sink

A two-phase commit sink writes a streaming job’s output inside a transaction that is prepared as records are processed and committed only when the corresponding checkpoint has succeeded. If the job fails before that point, the transaction is aborted and the replayed records write a fresh one — so the destination never holds output from an attempt that was later rolled back.

It exists because an engine’s strongest guarantee stops at its own boundary. Flink’s phrasing is that exactly-once “means that every event will affect the state being managed by Flink exactly once,” and the documented conditions for extending that end to end are that “your sources must be replayable, and your sinks must be transactional (or idempotent).” Descriptions below follow Apache Flink and Apache Kafka documentation, checked in September 2026.

The two phases

1. Prepare. As it processes, the sink writes to the destination in a way that is durable but not yet visible to readers — an open transaction, a staging location, an uncommitted batch. Nothing is lost if the process dies here, and nothing is exposed either.
2. Commit. When the checkpoint that covers those records completes, the sink commits. The write becomes visible, and it becomes visible at the same logical point as the engine’s own recorded progress.

Tying the commit to the checkpoint is the whole trick. The checkpoint records where the input had been read to; the commit makes the corresponding output visible. What the pairing buys is that a restart never has to guess which of the two happened.

It does not mean the two are one atomic action. A process can die after the checkpoint completes and before the destination commit returns, and that window is exactly why the pattern has a recovery half. The rule that makes it correct is which side the checkpoint is the record of truth for: a transaction belonging to a completed checkpoint must be committed on recovery, and one belonging to an incomplete checkpoint must be aborted.

• Failure before the checkpoint completes. The staged output is abandoned — aborted, or left never made visible — and the input is reprocessed from the previous position.
• Failure after it completes, before or during the commit. The job restarts from that checkpoint and re-commits the pending transaction. The commit therefore has to be repeatable: committing something already committed must succeed rather than error, since the previous attempt may have landed.

Three requirements follow, and they fail in different ways rather than all degrading to at-least-once.

• The transaction identifier must be in the checkpoint. A restarted process has to name the transaction it is resuming. Without it, recovery cannot find the pending work at all, and whether that ends in a duplicate or a gap is luck.
• The destination must keep the transaction alive across the restart. Every system has a limit, and this is the requirement whose failure is loss, not duplication: the completed checkpoint already advanced the input position, so if the transaction expires or is aborted by the destination, restarting from that checkpoint reprocesses nothing and the output for those records never exists. Reprocessing is not automatic here — it has to be arranged.
• An unresolvable pending transaction must stop the job. Proceeding past one is how the same output is committed twice, or silently skipped.

The second case deserves a plan rather than a retry, because the two sides have to be moved back together: restart from an earlier checkpoint or savepoint whose input position precedes the lost output, and reconcile the destination for that range first — remove whatever the abandoned transaction may have left visible, or make the re-run replace the range rather than add to it. Choosing an outage window longer than the destination’s transaction timeout is therefore a decision to accept manual recovery, which is worth knowing before the outage rather than during it.

Note also what the commit does not tell you. A commit that times out or loses its response has an unknown outcome — it may have landed — which is precisely why recovery re-commits by identifier and why that re-commit must succeed rather than error when the work is already committed. “The first attempt is never committed” is the intended behaviour of the prepare phase, not a guarantee about an in-flight commit.

One consequence worth naming for anything with a visible side effect: the pattern bounds duplicates in the destination store, not in the world. Moving an action to commit time reduces how often it fires spuriously; it does not make it exactly-once, because the commit itself can be retried and the action re-executed. A notification, a payment, or an outbound call therefore still needs its own de-duplication, keyed on something stable — the transaction identifier or a business key — rather than on the attempt.

Transactional or idempotent: choosing

	Transactional sink	Idempotent sink
How a repeat is made harmless	Staged output is not visible until commit, and a re-commit by identifier is a no-op	The repeat overwrites the same keyed row, changing nothing
Needs from the destination	Transactions, or an atomic way to make staged output visible	A stable key and an update-in-place write
Cost	Output appears only at checkpoint boundaries, so latency follows the checkpoint interval	None structurally, but every write must carry the key and the destination must honor it
Fails when	The destination has no transaction and no atomic swap	The destination cannot recognise a repeat — no stable key to write against, or a receiver that treats every request as new

Where the destination is a table format or an object store, the atomic-visibility half of this is the same mechanism as atomic publication: stage the files, then move one pointer. The streaming addition is only that the pointer moves when the checkpoint says so.

The reader has to cooperate

A transaction protects nothing if the consumer reads uncommitted data, and this is the step most often skipped. Kafka makes the setting explicit: with isolation.level set to read_committed, a consumer “will only return transactional messages which have been committed,” while the default read_uncommitted “will return all messages, even transactional messages which have been aborted.”

The correct setting has a visible cost, which is better anticipated than discovered. In read_committed mode a consumer reads “only … messages up to the last stable offset (LSO), which is the one less than the offset of the first open transaction” — so while a transaction is in flight, the consumer cannot read to the end of the log. Correctness of this kind is paid for in latency, and the amount is roughly the producer’s commit interval.

What it cannot cover

Some effects have no transaction to join. An email that has been sent is sent; a payment request without an idempotency key charges twice; a third-party API that acknowledges and then times out leaves the caller unable to tell what happened.

Only one of the usual three responses actually removes the duplicate. Giving the effect a de-duplication key the receiver honours — a payment idempotency key, a message id the downstream system rejects on repeat — does; the receiver does not have to be a table for this, only able to recognise a repeat. Deferring the effect until after the commit changes when it fires and how often it fires spuriously, but a retried commit re-executes it, so it is a reduction in exposure rather than a fix. And accepting occasional duplication is a decision to detect and correct it, which is why a periodic reconciliation stays part of the design rather than being an admission that it failed.

So the question to ask of a side effect is not whether it can wait for the commit, but whether the thing receiving it can tell that it has seen this one before.

How sinks fit with sources, state, and the rest of end-to-end correctness is worked through in Stream Processing Semantics; the state side is covered in stateful stream processing.

References: Apache Flink Documentation, Learn Flink: Fault Tolerance via State Snapshots; Apache Kafka Documentation, Consumer Configs (isolation.level).