End-to-End Exactly-Once with Apache Flink: Two-Phase Commit Sink and Checkpoints

Achieving true End-to-End Exactly-Once Processing (EOS) means that from the moment an event is ingested from an upstream source (Kafka), through distributed stateful stream operators (Flink), to the final write in an external database or search index (PostgreSQL, ClickHouse, Kafka), every record affects final state exactly once—even in the presence of machine crashes, network partitions, and JVM failures.

To achieve this without degrading real-time streaming throughput, Apache Flink pairs the Chandy-Lamport Distributed Snapshot Algorithm with its TwoPhaseCommitSinkFunction.

1. The Chandy-Lamport Checkpointing Algorithm

Instead of pausing the entire cluster to take a consistent state backup, Flink injects lightweight Checkpoint Barriers directly into the event stream at the sources:

The Magic of Stream Alignment:

• Barriers divide the infinite stream into two disjoint epochs: Events before the barrier (included in Checkpoint $N$) and Events after the barrier (included in Checkpoint $N+1$).
• Stream records continue processing uninterrupted in RAM while state snapshots are asynchronously written to remote object storage (S3 / GCS).

2. The TwoPhaseCommitSinkFunction Interface

To extend Flink's internal exactly-once guarantees to external downstream systems, sink connectors implement four lifecycle methods:

3. Detailed Step-by-Step Execution Walkthrough

1. beginTransaction(): When a new checkpoint cycle starts, the sink operator initiates a new transaction $T_1$ in the external datastore (e.g., calling BEGIN in PostgreSQL or starting a transactional producer in Kafka).
2. invoke(value): All records arriving before the next checkpoint barrier are written speculatively into transaction $T_1$. Downstream users cannot see these uncommitted rows.
3. preCommit(): When Checkpoint Barrier $N$ arrives, the sink flushes its network buffers, closes $T_1$ to further writes, and records $T_1$'s transaction handle in Flink's state store. It immediately spawns transaction $T_2$ for upcoming events.
4. commit(): Once the Flink JobManager confirms that all operators in the pipeline successfully completed Checkpoint $N$, it issues a global commit notification. The sink executes COMMIT on transaction $T_1$.
5. Recovery on Failure: If a worker crashes before Checkpoint $N$ completes, the JobManager restores all operators to Checkpoint $N-1$. The restored sink calls abort() on all pending speculative transactions, eliminating partial writes.

4. Multi-Language Production Example: Java Flink Exactly-Once Kafka Sink

5. Summary & Operational Gotchas

1. Transaction Timeout Matching: Ensure the downstream database or Kafka transaction timeout (transaction.timeout.ms) is significantly larger than Flink's maximum checkpoint timeout (checkpointTimeout + 5 minutes). If Kafka aborts the transaction before Flink completes the checkpoint, the pipeline will fail.
2. PostgreSQL XA Limits: When using two-phase commit sinks with PostgreSQL, ensure max_prepared_transactions is set appropriately to accommodate concurrent checkpointing tasks.
3. Storage Overhead: Uncommitted transactional records in Kafka or databases occupy space on disk until committed; ensure adequate disk provisioning.