A Flink checkpoint is one global state: a position in every source plus the state of every operator, as if everything before the barrier had been processed and nothing after it. “Exactly-once” is a property of that state. An external effect falls inside the guarantee only if it is committed atomically with the checkpoint (a transactional sink) or is idempotent (22). Everything else — an HTTP call, a non-transactional write, an alert — is replayed from the last completed checkpoint on recovery, which makes it at-least-once, whatever the job config says.
Mechanism. Chandy–Lamport assume FIFO, lossless channels: a marker travels along every channel; a process records its own state on the first marker, and a channel’s state is whatever arrived on it after that and before its own marker. The theorem is subtle: the recorded state is reachable from the start and the final state is reachable from it, yet it may never have existed in the actual run — a checkpoint is a possible state, not a photo. Flink’s asynchronous barrier snapshotting (Carbone et al.) aligns barriers on acyclic graphs so channel state is empty and only operator state is stored. Alignment blocks the inputs that already delivered a barrier, so under backpressure checkpoint duration tracks the queue (23). Unaligned checkpoints let the barrier overtake buffers and persist the in-flight data — Chandy–Lamport channel state is back, checkpoints grow, and on recovery watermarks are emitted only after the in-flight data is restored. aligned-checkpoint-timeout starts aligned and switches only when alignment is slow.
Where the cut meets Kafka. With EXACTLY_ONCE, KafkaSink writes into a Kafka transaction that is committed after the checkpoint completes. Four consequences are routinely missed. (1) The consumer default is isolation.level=read_uncommitted: such a consumer also reads records of aborted transactions, so the guarantee silently stops at the topic. (2) A read_committed consumer only reads up to the last stable offset, so output becomes visible in checkpoint-sized steps — worst case the interval plus the checkpoint duration — and a slow checkpoint stalls every such reader of the partition. (3) If the broker aborts the transaction before Flink commits it — transaction.timeout.ms shorter than the longest checkpoint plus the longest restart — the data is lost; the sink builder defaults that timeout to one hour, above the broker’s default cap transaction.max.timeout.ms of 15 minutes, which fails at producer initialization until one of the two is changed. (4) The offsets KafkaSource commits to the consumer group are for monitoring only; recovery uses the offsets inside the checkpoint. The same boundary drives replay (80) and state evolution (82); end-to-end deduplication is still the last word (21).
Configuration conf = new Configuration();
conf.set(CheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(30));
conf.set(CheckpointingOptions.CHECKPOINTING_CONSISTENCY_MODE, CheckpointingMode.EXACTLY_ONCE);
conf.set(CheckpointingOptions.CHECKPOINTING_TIMEOUT, Duration.ofMinutes(5));
// start aligned; if alignment takes longer than 10 s (backpressure), store in-flight buffers
conf.set(CheckpointingOptions.ENABLE_UNALIGNED, true);
conf.set(CheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT, Duration.ofSeconds(10));
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
KafkaSink<Order> sink = KafkaSink.<Order>builder()
.setBootstrapServers(brokers)
.setRecordSerializer(serializer)
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
// stable across restarts, unique among all jobs writing to this cluster
.setTransactionalIdPrefix("orders-enricher")
// must exceed checkpoint timeout (5 min) + worst restart, and must not exceed the
// broker's transaction.max.timeout.ms (default 15 min); the builder default is 1 h
.setProperty("transaction.timeout.ms", String.valueOf(Duration.ofMinutes(15).toMillis()))
.build();
// Outside this job: every consumer of the output topic needs
// isolation.level=read_committed (the default is read_uncommitted), and sees
// new records only after the checkpoint that contains them completes.