Definition
- Internal exactly-once – Barrier-based checkpointing snapshots all operator state and source offsets so that on recovery, state reflects each record exactly once.
- End-to-end exactly-once – Extends internal guarantees to external sinks by tying the sink's transaction to the checkpoint lifecycle.
- Two-phase commit – A protocol with four hooks (beginTransaction, preCommit, commit, abort) that ensures the sink only makes output visible when the checkpoint is confirmed complete.
- Pre-commit – The sink flushes buffered output into an open transaction and prepares it when the checkpoint barrier arrives, but does not commit.
- Commit – Runs only after the JobManager confirms the checkpoint is globally complete, making output visible to downstream consumers.
- Abort – Rolls back a transaction that will not be committed, for example on failure before checkpoint completion.
The National Bank of India wants to guarantee that every transaction is written to the external ledger exactly once, never zero times and never twice, even if a server crashes mid-write. The trick is a two-stage signature. When the clerk reaches a checkpoint, they prepare the ledger entries but leave them in a sealed envelope that is not yet filed (pre-commit). Only once head office confirms that the whole checkpoint across all clerks succeeded does the clerk open the envelope and file the entries for real (commit). If a crash happens before head office confirms, the sealed envelope is simply torn up and recovery re-does the work from the last good checkpoint.
Understanding the Concepts
What is internal exactly-once in Flink?
Internally, exactly-once state is guaranteed by checkpointing. Barriers flow through the graph, operators snapshot a consistent cut, and sources record their offsets. On recovery, Flink restores state and rewinds sources so each record updates state exactly once.
package com.bank.streaming;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
public class InternalExactlyOnceDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
// Enable checkpointing for internal exactly-once
env.enableCheckpointing(30_000); // every 30 seconds
env.getCheckpointConfig().setCheckpointingMode(
CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointStorage(
"s3://nbi-checkpoints/flink/");
env.addSource(new KafkaSource<>("nbi-transactions"))
.keyBy(Transaction::getAccountId)
.process(new AccountBalanceTracker()) // keyed state
.print();
// On failure: restore state + rewind Kafka offsets
// Each record updates state exactly once
env.execute("NBI Internal Exactly-Once Demo");
}
}
Why is internal exactly-once not enough?
Internal consistency means operator state is correct, but the sink writes to the outside world. If the sink wrote output as it processed and the job then failed and replayed, the replayed records would be written again, producing duplicates in the external system.
Problem without transactional sink:
1. Flink processes records R1, R2, R3
2. Sink writes R1, R2, R3 to external database
3. Crash before checkpoint completes
4. Recovery: Flink replays from last checkpoint
5. Sink writes R1, R2, R3 AGAIN
6. External database now has R1, R2, R3 TWICE
Solution: sink participates in checkpoint protocol
How does the two-phase commit protocol work?
The two-phase commit ties the external transaction to the checkpoint lifecycle. It has four hooks: beginTransaction opens a new transaction, preCommit prepares it when the barrier arrives, commit runs only after checkpoint confirmation, and abort rolls back on failure.
package com.bank.streaming;
import org.apache.flink.streaming.api.functions.sink
.TwoPhaseCommitSinkFunction;
public class ExactlyOnceLedgerSink
extends TwoPhaseCommitSinkFunction<Transaction,
JdbcTransaction, Void> {
@Override
protected JdbcTransaction beginTransaction() {
// Open new JDBC transaction
Connection conn = dataSource.getConnection();
conn.setAutoCommit(false);
return new JdbcTransaction(conn);
}
@Override
protected void invoke(JdbcTransaction txn, Transaction record,
Context context) throws Exception {
// Buffer writes into the open transaction
PreparedStatement ps = txn.getConnection().prepareStatement(
"INSERT INTO nbi_ledger (account_id, amount, txn_time) "
+ "VALUES (?, ?, ?)");
ps.setString(1, record.getAccountId());
ps.setBigDecimal(2, record.getAmount());
ps.setTimestamp(3, record.getTxnTime());
ps.executeUpdate();
}
@Override
protected void preCommit(JdbcTransaction txn) {
// Flush and prepare but do NOT commit
// Called when checkpoint barrier reaches the sink
txn.getConnection().flush();
}
@Override
protected void commit(JdbcTransaction txn) {
// Called ONLY when checkpoint is globally confirmed
txn.getConnection().commit();
txn.getConnection().close();
}
@Override
protected void abort(JdbcTransaction txn) {
// Rollback on failure
txn.getConnection().rollback();
txn.getConnection().close();
}
}
How does the Kafka sink achieve exactly-once?
The Kafka sink uses Kafka's transactional producer. Pre-committed records are invisible to consumers reading in read_committed mode until the commit lands. The transactional.id prefix links the Flink checkpoint to the Kafka transaction.
package com.bank.streaming;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.connector.base.DeliveryGuarantee;
public class KafkaExactlyOnceDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(30_000);
KafkaSink<Transaction> sink = KafkaSink.<Transaction>builder()
.setBootstrapServers("kafka:9092")
.setRecordSerializer(
KafkaRecordSerializationSchema.builder()
.setTopic("nbi-settlements")
.setValueSerializationSchema(
new TransactionSerializer())
.build())
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
.setTransactionalIdPrefix("nbi-settlement-sink")
.build();
env.addSource(new KafkaSource<>("nbi-transactions"))
.keyBy(Transaction::getAccountId)
.process(new SettlementCalculator())
.sinkTo(sink);
env.execute("NBI Kafka Exactly-Once Demo");
}
}
What is the visibility latency with exactly-once sinks?
Downstream readers only see output after the checkpoint completes and the commit lands. The checkpoint interval becomes the end-to-end latency floor for exactly-once output.
Visibility timeline:
t=0s Checkpoint interval starts
t=0-30s Sink processes records into open Kafka transaction
Records INVISIBLE to read_committed consumers
t=30s Checkpoint barrier reaches sink --> preCommit
t=31s Checkpoint confirmed by JobManager --> commit
Records NOW VISIBLE to consumers
Visibility latency = checkpoint interval + checkpoint duration
With 30s interval: consumers see data ~30-35s after production
// Trade-off: lower checkpoint interval = lower latency but more overhead
env.enableCheckpointing(10_000); // 10s: lower latency, more checkpoints
env.enableCheckpointing(60_000); // 60s: higher latency, fewer checkpoints
// Choose based on business latency requirements vs checkpoint cost
What about idempotent sinks as an alternative?
If the external system cannot do transactions, you can achieve effectively-once with an idempotent sink that uses a deterministic ID. Replays overwrite rather than duplicate.
package com.bank.streaming;
import java.math.BigDecimal;
public class IdempotentSinkDemo {
// Deterministic ID based on record content
// Same record always produces same ID
public static String deterministicId(Transaction txn) {
return txn.getAccountId() + "-" + txn.getTxnTime().getTime()
+ "-" + txn.getAmount().toPlainString();
}
// Idempotent upsert: INSERT ... ON CONFLICT UPDATE
// Replays overwrite the same row, no duplicates
// "INSERT INTO nbi_ledger (txn_id, account_id, amount) "
// "VALUES (?, ?, ?) "
// "ON CONFLICT (txn_id) DO UPDATE SET amount = EXCLUDED.amount"
}
What are common mistakes with exactly-once sinks?
The most common mistake is not setting the Kafka transaction.timeout.ms higher than the maximum checkpoint interval plus recovery time. If the Kafka transaction expires before Flink can commit it after recovery, the data is lost.
// WRONG: default Kafka transaction timeout (15 min)
// Checkpoint interval: 5 min, recovery time: up to 20 min
// If recovery takes 20 min, transaction expires before commit!
// CORRECT: set transaction timeout > checkpoint interval + recovery
KafkaSink.<Transaction>builder()
.setProperty("transaction.timeout.ms", "1800000") // 30 min
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
.setTransactionalIdPrefix("nbi-settlement-sink")
.build();
// Also ensure Kafka broker: transaction.max.timeout.ms >= 1800000
What should a senior engineer know about exactly-once?
A senior engineer should know that exactly-once is a contract across the whole pipeline: a source that cannot replay (non-replayable) or a sink that cannot transact or deduplicate breaks it regardless of Flink's internal guarantees. They should also know the end-to-end latency trade-off and the transaction timeout pitfall.
Exactly-once requirements:
Source: Must be replayable (Kafka, Kinesis, files)
State: Checkpointed with EXACTLY_ONCE mode
Sink: Must be transactional (2PC) or idempotent
If ANY component breaks the chain:
Source not replayable? --> at-most-once possible
Sink not transactional? --> at-least-once with duplicates
Checkpointing disabled? --> no recovery guarantees
package com.bank.streaming;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
public class SeniorExactlyOnceConfig {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
// Complete exactly-once configuration
env.enableCheckpointing(30_000);
CheckpointConfig cfg = env.getCheckpointConfig();
cfg.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
cfg.setMinPauseBetweenCheckpoints(10_000);
cfg.setCheckpointTimeout(120_000);
cfg.setCheckpointStorage("s3://nbi-checkpoints/");
cfg.setExternalizedCheckpointCleanup(
CheckpointConfig.ExternalizedCheckpointCleanup
.RETAIN_ON_CANCELLATION);
// Source: Kafka (replayable)
// Sink: Kafka with EXACTLY_ONCE + transactional ID
// Full end-to-end exactly-once chain
env.execute("NBI Senior Exactly-Once Config");
}
}
Visual Overview
End-to-End Exactly-Once with Two-Phase Commit
Checkpoint Lifecycle:
JobManager Source Operator Sink
│ │ │ │
│──trigger checkpoint──▶│ │ │
│ │──barrier──▶ │ │
│ │ snapshot │──barrier──▶│
│ │ (offsets) │ snapshot │
│ │ │ (state) │ preCommit
│ │ │ │ (flush to txn)
│◀── ack ───────────────│◀── ack ────│◀── ack ────│
│ │
│──checkpoint complete──────────────────────────────▶│
│ │ COMMIT
│ │ (output visible)
Failure Before Checkpoint Confirmation:
│──trigger checkpoint──▶│──barrier──▶│──barrier──▶│
│ │ │ │ preCommit
│ X CRASH │ │
│ │
│ Checkpoint NOT confirmed │
│ Sink: ABORT transaction │
│ Recovery: restore last good checkpoint │
│ Replay from saved offsets │
│ No duplicate output (transaction was aborted) │
Kafka Exactly-Once Visibility:
Producer (Flink Sink) Kafka Topic Consumer
───────────────────── ─────────── ────────
beginTransaction │ │
write R1, R2, R3 │ R1,R2,R3 (uncommit) │
preCommit │ │ (invisible)
... checkpoint confirmed ... │ │
COMMIT │ R1,R2,R3 (committed)│
│ │ NOW VISIBLE
Key Takeaway
End-to-end exactly-once in Flink means every input record affects the final external output exactly once, even across failures. It is achieved by combining two mechanisms. The first is barrier-based checkpointing, which gives exactly-once state consistency internally: barriers flow through the graph, operators snapshot a consistent cut, and on recovery Flink restores state and rewinds sources. The second is the two-phase commit protocol for the sink, which ties the external transaction to the checkpoint lifecycle.
The protocol has four hooks. beginTransaction opens a new external transaction when a checkpoint interval starts. preCommit runs when the checkpoint barrier reaches the sink: it flushes buffered output into the transaction and prepares it, but does not commit. commit runs only after the JobManager confirms the checkpoint is globally complete, making output visible. abort rolls back a transaction on failure. Because the commit happens only on checkpoint completion, any failure before that leaves the transaction uncommitted.
At the National Bank of India we used this for our settlement ledger pipeline. The Kafka sink used Kafka transactions with a transactional ID prefix, and the accounting consumers read in read-committed mode so they only saw settlements after the commit. When I simulated a crash after pre-commit but before confirmation, the sink aborted the uncommitted transaction, recovery replayed the same settlements, and they ended up committed exactly once with no duplicates.
The key senior points are that the checkpoint interval becomes the latency floor for exactly-once output, that the Kafka transaction timeout must exceed your checkpoint interval plus recovery time, and that if the sink cannot do transactions you can fall back to idempotent writes.
Remember one sentence: End-to-end exactly-once pairs checkpointed state with a two-phase-commit sink that pre-commits on barrier and commits only when the checkpoint is globally confirmed.
---