Why does end-to-end exactly-once in Flink require a two-phase-commit sink?

## Where the guarantee stops A completed Flink checkpoint means: every operator's state and every source's position correspond to one consistent cut of the input. If a task fails, Flink restores that state and replays from that cut. Everything inside the job is therefore exactly-once — each input record is *reflected in state* exactly once, no matter how many times it is physically processed. Nothing about this constrains the outside world. If the sink wrote row 4,001 to the database and the job then fails back to a checkpoint taken at record 4,000, the row is still in the database and the replay writes it again. The engine has no way to un-send a network call. So end-to-end exactly-once is a property of the *sink*, and Flink's job is only to give the sink a trustworthy signal about when a cut has become permanent. ## The two-phase-commit contract Flink gives sinks two hooks that map onto a two-phase commit: - **Snapshot** (barrier passes the sink task): the sink flushes everything written so far into a durable but not-yet-visible form, and stores the *handle* to that pending transaction in the checkpoint state. This is the vote-to-commit phase. - **Completion notification** (the coordinator confirms every subtask acknowledged): the sink commits the pending transaction. This is the commit phase, and it is the first instant at which the job knows the snapshot will never be rolled back. In Flink 2.3 these hooks belong to the **Sink V2** API in `org.apache.flink.api.connector.sink2`. A sink that implements `SupportsCommitter` is built from a `CommittingSinkWriter`, whose `prepareCommit()` turns the open transaction into *committables* at the checkpoint, and a `Committer`, whose `commit(...)` makes them visible once the checkpoint completes. The committables are checkpointed on their way to the committer, which is how a restarted job finds them again. The older `TwoPhaseCommitSinkFunction` — `beginTransaction`, `preCommit`, `commit`, `abort`, `recoverAndCommit` — was built on `SinkFunction`, which 2.0 removed from the public API; only a deprecated internal copy survives under `org.apache.flink.streaming.api.functions.sink.legacy`, so new sinks are written against Sink V2. Different code, identical contract. ## Walking the failure cases The contract only earns its keep if it is correct at every crash point: 1. **Crash before pre-commit.** The transaction was never made durable; it is aborted or abandoned. Recovery replays those records and writes them into a fresh transaction. No duplicate, no loss. 2. **Crash between pre-commit and commit.** The checkpoint may or may not have completed. If it completed, the transaction handle was in it: recovery restores the committable and commits it — a second time, if the crash came just after the commit — which is why the Sink V2 `Committer` contract requires commits to be idempotent: a committable that was already committed must leave the external system unchanged and is reported with `signalAlreadyCommitted()`. If the checkpoint did not complete, the job falls back to an earlier one and the pending transaction is aborted. 3. **Crash after commit.** The checkpoint completed and the data is visible; recovery starts from a cut at or after it, so nothing is rewritten. That is the whole argument, and being able to walk it is what a senior interview is testing. ## Kafka as the worked example ```java KafkaSink<String> sink = KafkaSink.<String>builder() .setBootstrapServers("broker:9092") .setRecordSerializer(/* ... */) .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) .setTransactionalIdPrefix("orders-out-") .build(); ``` Three things fall out of this that candidates routinely miss: - **Consumers must read committed.** Records inside an uncommitted Kafka transaction are on the broker but invisible to a consumer configured for committed reads. A downstream consumer left on the default isolation will see aborted output and the whole exercise is wasted. - **The transaction timeout must exceed the checkpoint interval.** `KafkaSink` opens a fresh transaction right after each checkpoint's snapshot and commits it only when the *next* checkpoint completes, so every transaction lives for roughly a full checkpoint interval, plus any downtime before recovery commits it. Unless you override it, the builder sets the producer's `transaction.timeout.ms` to one hour. If the producer's transaction timeout elapses first, the broker aborts it and that data is *lost*, not duplicated. The producer's timeout must also stay under the broker's maximum, so a long checkpoint interval forces you to raise both. - **The transactional id prefix must be stable and unique per job.** Two jobs sharing a prefix will fence each other's producers. `FileSink` does the same dance without a transaction manager: records go to in-progress part files, a part file that rolls becomes *pending* (bulk formats roll at every checkpoint), and when the checkpoint completes the committer renames pending files to their final visible names. ## The latency you are buying Output becomes visible only when a checkpoint completes, so end-to-end latency is at least the checkpoint interval. A job checkpointing every five minutes with a transactional sink is a five-minute-latency pipeline no matter how fast its operators run. This is the single biggest practical objection to exactly-once sinks and the reason many teams choose an idempotent sink instead. ## The alternative If the sink is naturally idempotent — an upsert keyed by a deterministic business key, an overwrite of a partition, a write whose key encodes the record's identity — then replayed records overwrite rather than duplicate and the *final result* is correct without any transaction. That is at-least-once processing with an idempotent sink, and for many pipelines it is both cheaper and lower-latency than two-phase commit.