Two-Phase Commit and Exactly-Once Sinks

Exactly-once processing is usually described as a property of the stream processor, and it is not one. A processor can guarantee that its own state reflects each input record once, because its state is under its control and can be rolled back to a checkpoint. The moment an operator writes to an external system, the guarantee depends on that system’s ability to accept a write tentatively and make it visible later. Without that, the choice is at-least-once with duplicates or at-most-once with loss, and no amount of internal checkpointing changes it.

Where the duplicate comes from

A job fails after writing to a sink and before its next checkpoint completes. On restart it resumes from the last checkpoint and reprocesses the records after it, including the one already written. The write happened, the state does not know it happened, and the record is emitted twice.

The naive fix is to make the sink idempotent — key each write by a deterministic identifier and let the sink overwrite. That works when the sink has a natural primary key and the operation is an upsert, and it fails for anything append-only. Appending a row to a table or a message to a topic has no natural key unless the writer invents one, and even with an invented key, a sink that does not enforce uniqueness will accept the duplicate.

The stronger fix is to make the write atomic with the checkpoint: the write is invisible until the checkpoint that follows it is complete, and it is discarded if the job fails before that. This requires the sink to support a transaction with a prepare phase, which is what makes the protocol a two-phase commit rather than a retry loop.

The protocol, in the sink’s terms

A sink participating in checkpointing maintains a transaction that spans the interval between two checkpoints. On each checkpoint, the sink’s task does three things in order: it flushes whatever it has written into the transaction, it reports the transaction’s identity to the checkpoint coordinator as part of the snapshot, and it does not commit. The commit happens only after every task in the job has reported its snapshot successfully.

The two phases are therefore: prepare, where each sink flushes and hands over its transaction handle; and commit, where the coordinator, having collected all handles, instructs each sink to commit. If any task fails before the coordinator has all handles, the checkpoint is abandoned and the sinks are instructed to abort their open transactions. Recovery restores state from the previous complete checkpoint and reopens transactions from there.

The sink’s transaction lifecycle has four operations, and the checkpoint protocol calls them in a fixed order:

Operation	Called when	Effect
begin	the first record of a new transaction	opens a transaction, returns its handle
pre-commit	the checkpoint that covers those records is triggered	flushes the writes; the handle enters the snapshot
commit	every task’s snapshot succeeded	makes the writes visible atomically
abort	the checkpoint was abandoned	discards the writes

The important structural detail is that commit is driven by a notification from the checkpoint coordinator, not by the sink’s own timer. A sink that commits on a schedule independently of the checkpoint has reintroduced the gap: it can commit writes whose checkpoint never completed.

What the sink must actually provide

Three properties are required, and a sink that lacks any one of them cannot give the guarantee.

The first is durability of the tentative write. Records written in the prepare phase must survive a crash of the sink’s own process, or the commit after recovery has nothing to commit. A sink that buffers in memory until commit loses the prepared data and must replay it, which returns the pipeline to at-least-once.

The second is atomic visibility across all partitions written by the job. If a job writes to several partitions of a topic, the commit must make all of them visible together, or a consumer reading across partitions sees a partial result. This is the requirement that makes the protocol expensive: the sink must maintain a commit marker that readers consult, and readers must be aware of it.

The third is abort. A transaction left open by a failed attempt must be discarded, and the discard must be safe even if it races with a commit instruction for a different checkpoint. Sinks that cannot abort accumulate open transactions, and the accumulated state is both a resource leak and a correctness hazard.

A transactional message broker satisfies all three by writing records into the log with a transaction marker, so a consumer that reads with the right isolation level sees either all records of a transaction or none. A filesystem sink can satisfy them by writing to a temporary path and renaming atomically, provided the rename is atomic and the reader only lists final paths. A database sink satisfies them if it has real transactions and the job’s writes fit within one.

The cost: latency tied to the checkpoint interval

The commit happens at checkpoint boundaries, so the sink’s visibility lags the processing by up to one checkpoint interval. A job checkpointing every ten seconds emits results that become visible in ten-second steps, which is fine for a warehouse load and unacceptable for a dashboard that expects sub-second freshness.

Reducing the interval reduces the lag and increases the snapshot overhead, which is the state-size problem from Keyed State Layout and the Cost of Keeping It. The two costs pull in opposite directions, and the operating point is usually set by how much recovery time the downstream can tolerate rather than by the freshness requirement, because recovery time is the harder constraint.

A separate cost is that a long-running transaction holds resources in the sink. A message broker retains the open transaction’s records and delays the advancement of the consumer-visible offset until commit, so readers with read-committed isolation stall behind the longest open transaction. A job that checkpoints infrequently therefore imposes a delay on every consumer of its output, not just on its own recovery.

Where exactly-once ends

The guarantee is bounded by the set of systems that participate in it. A pipeline that reads from a replayable source, checkpoints its state, and writes transactionally to one sink has exactly-once semantics end to end. Add a side effect that does not participate — a call to an external service, a write to a log file, an email — and that side effect is at-least-once regardless of the rest.

This is not a theoretical caveat. The common shape of a streaming job includes a notification step, a metrics emission, or a cache invalidation, and each of those is outside the transaction. The honest description of such a job is that its data output is exactly-once and its side effects are at-least-once, and the design work is to make the side effects idempotent or to move them behind a deduplication step keyed by the output record’s identity.

The second boundary is the source. Exactly-once requires that the source can be replayed from a checkpointed position, which means the source must retain its data for at least the recovery window. A source with a short retention, or one that discards records on acknowledgement, can force the job to resume past data it needed. The retention requirement is set by the recovery time, which is set by the state size and the checkpoint interval — the same three quantities appear in every part of this design.

Reading and writing the same system

When the source and sink are the same system, the transaction can span both, and the read position becomes part of the transaction rather than a separate checkpointed value. This removes one failure mode — a committed write paired with an uncommitted read offset — at the cost of coupling the job’s recovery to the broker’s transaction timeout. A transaction that stays open longer than the broker’s configured maximum is aborted by the broker, and the job’s next checkpoint fails with a message about an expired transaction rather than about its own state.

The timeout is a hard limit that the checkpoint interval must stay under, and it is not always visible from the job’s configuration. A pipeline that works with a one-minute checkpoint interval and fails after being changed to five minutes is usually hitting this limit, and the diagnosis requires knowing that the limit exists.

The result is a guarantee that is real but conditional, and the conditions are the interesting part of any claim about exactly-once delivery. The internal state is the easy half; the external commitment is where the semantics are decided, and it is decided by the sink’s capabilities rather than by the processor’s.