A stream processor that crashes will reprocess some records. There is no way around it: the work done between the last durable checkpoint and the crash is lost, and the only way to recover it is to read those records again. Exactly-once stream processing does not prevent that replay. It makes the replay harmless, so that the state the job keeps and the output it publishes reflect every input record once.
This article is about the architecture that achieves that end to end. It treats exactly-once as a chain of three links: a source that can rewind, a snapshot that captures state and source position together, and an output protocol that ties visibility to that snapshot. We look at four ways to build the last link, map the same chain onto Flink, Kafka Streams and Spark, inject a crash at every step of a worked pipeline, and list what still cannot be made exactly-once.
What exactly-once actually guarantees
Be precise about the claim. A record may be read, transformed and even written several times. What is guaranteed is that the committed result is the same as if each record had been processed exactly once: counters are not double-counted, and downstream readers see each output record once. Some people call this effectively-once, which is the more accurate name.
The guarantee covers state the framework manages and outputs that join its protocol. It does not cover an email sent, an HTTP call made or a metric incremented from inside an operator. Those happen whenever the code runs, including during replay, and need their own idempotency.
What exactly-once actually guarantees
Be precise about the claim. A record may be read, transformed and even written several times. What is guaranteed is that the committed result is the same as if each record had been processed exactly once: counters are not double-counted, and downstream readers see each output record once. Some people call this effectively-once, which is the more accurate name.
The guarantee covers state the framework manages and outputs that join its protocol. It does not cover an email sent, an HTTP call made or a metric incremented from inside an operator. Those happen whenever the code runs, including during replay, and need their own idempotency.
The three-link chain
Each link fails differently if it is missing. Without a rewindable source, replay has nothing to read and records are lost. Without a consistent snapshot, recovered state and source position disagree, so some records are counted twice or not at all. Without a committed output, state is right but downstream systems see the duplicate writes made during replay. The guarantee is only as strong as the weakest link, and most production exactly-once bugs are one link quietly configured away.
Link one: a source that can rewind
A Kafka topic is the canonical replayable source: records stay for the retention period and can be read again from any offset. The key architectural rule is that the authoritative position is the offset stored in the processor's own snapshot, not the consumer-group offset committed to Kafka. On recovery the job seeks every partition to the snapshotted offset. Committed group offsets are still useful for lag monitoring, but they are not used for correctness. Retention must comfortably exceed the longest plausible outage plus recovery, or a long incident turns into data loss.
Sources that cannot rewind, such as webhooks, sockets or a message queue that deletes on acknowledgement, cannot join the chain directly. Land them in a durable log first, acknowledging the sender only after the log write succeeds, and let the stream job read from the log. Change data capture follows the same rule: the database's replication log is the replayable source, and the connector's position in it belongs in the snapshot.
Link two: consistent snapshots of state
Flink takes consistent snapshots with checkpoint barriers, a variant of the Chandy-Lamport algorithm. The coordinator asks every source to record its offsets and inject barrier n into its output. Barriers flow with the records. An operator with several inputs waits until barrier n has arrived on all of them, which is called alignment, then snapshots its state and forwards the barrier. Its snapshot therefore reflects exactly the records before the barrier on every input, which matches the recorded offsets. Large state is uploaded asynchronously and, with RocksDB, incrementally.
Alignment is the price. Under backpressure, barriers queue behind slow records and checkpoints take longer. Unaligned checkpoints, enabled with enableUnalignedCheckpoints() on the checkpoint configuration, let barriers overtake buffered records and store those in-flight buffers in the snapshot instead. They keep checkpoints fast under backpressure at the cost of larger snapshots.
Link three: four sink protocols
The output link has four common designs. All of them tie what downstream readers see to a completed snapshot; they differ in how.
| Sink protocol | How it works | Cost | Fits |
|---|---|---|---|
| Transactional (two-phase) | Open a transaction per checkpoint; pre-commit at the barrier; commit when the checkpoint completes | Output visible only once per checkpoint interval | Kafka topics, XA-capable databases |
| Idempotent upsert | Deterministic key per result; replay overwrites with the same value | Readers can briefly see older values during replay | Key-value stores, search indexes, JDBC upsert |
| Offsets-in-sink | Write results and source offsets in one database transaction; resume from the stored offsets | Couples the job to one database | Single relational sink |
| Staged files | Write files to a pending location; rename or register them on checkpoint completion | File count and commit latency | Data lakes, table formats |
For Kafka output, Flink's KafkaSink implements the transactional protocol. Records produced during a checkpoint epoch go into an open Kafka transaction; they are durable on the brokers but invisible to read_committed consumers. At the barrier the sink flushes and records the transaction in the checkpoint. When the coordinator reports the checkpoint complete, the sink commits. On recovery the sink commits any transaction recorded in the last completed checkpoint, which may already have been committed before the crash, and aborts transactions it opened after that point.
env.enableCheckpointing(30_000); // checkpoint every 30 s
env.getCheckpointConfig().enableUnalignedCheckpoints(); // optional: fast under backpressure
KafkaSink<String> sink = KafkaSink.<String>builder()
.setBootstrapServers("broker:9092")
.setRecordSerializer(KafkaRecordSerializationSchema.builder()
.setTopic("settlements")
.setValueSerializationSchema(new SimpleStringSchema())
.build())
.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
.setTransactionalIdPrefix("settlements-v1") // stable and unique per job
.setProperty("transaction.timeout.ms", "900000") // must outlast checkpoint + recovery
.build();The timeout line matters. A Kafka producer's transaction.timeout.ms defaults to 60 seconds, and the broker caps it with transaction.max.timeout.ms, 15 minutes by default. The transaction opens at the start of its epoch, so it must stay alive through the checkpoint interval, the checkpoint itself and, in the worst case, a restart. If a checkpoint completes, the job crashes before the sink commits, and recovery takes longer than the transaction timeout, the broker has already aborted the transaction and that epoch's output is lost even though the checkpoint says it was committed.
When the sink is a single relational database, offsets-in-sink is often simpler than any transaction protocol: store the source position in the same transaction as the results.
BEGIN;
UPDATE stream_progress SET batch_id = :batch_id
WHERE job = 'totals' AND batch_id < :batch_id;
-- 0 rows updated: this batch was already applied, so ROLLBACK and skip it
INSERT INTO merchant_totals (merchant_id, day, total_cents)
VALUES (:m, :d, :delta)
ON CONFLICT (merchant_id, day) DO UPDATE
SET total_cents = merchant_totals.total_cents + EXCLUDED.total_cents;
COMMIT;The same chain in Kafka Streams and Spark
Kafka Streams builds the same chain from Kafka alone. With processing.guarantee set to exactly_once_v2, each stream thread writes output records, changelog updates for its state stores and its consumer offsets in one Kafka transaction, so all three commit or abort together. The commit interval drops to 100 ms by default under this setting, which bounds how long output stays invisible.
Spark Structured Streaming checkpoints offsets and state per micro-batch. Its file sink records each batch's files in a metadata log so readers see each batch once. With foreachBatch you write to arbitrary systems, and the guarantee is at-least-once unless you use the batch id the callback receives to skip batches already applied, exactly as the SQL above does.
Worked example: a crash at every step
Take a settlement pipeline: Kafka payments in, per-merchant running totals in keyed state, settlement records out to a Kafka topic, checkpoints every 30 seconds. Checkpoint 41 has completed. During epoch 42 the job reads 120,000 payments. Now crash it at each step and follow the result.
| Crash point | Recovery | Visible result |
|---|---|---|
| Mid-epoch, before barrier 42 | Restore checkpoint 41 state and offsets; abort open transaction; replay 120,000 records | Downstream never saw epoch 42; replayed output commits once |
| After sink pre-commit, before checkpoint 42 completes | Checkpoint 42 incomplete, restore 41; abort pre-committed transaction | Same as above |
| After checkpoint 42 completes, before sink commits | Restore 42; sink commits the transaction recorded in it | Epoch 42 appears once, slightly late |
| After commit, before the next barrier | Restore 42; nothing to redo | No duplicates |
| After checkpoint 42 completes, before commit, and recovery takes 20 minutes with a 15-minute timeout | Broker already aborted the epoch 42 transaction; there is nothing left to commit | Epoch 42 output lost: an exactly-once violation |
Only the last row breaks the guarantee, and it is a configuration error rather than a protocol flaw. It is also the case to rehearse: kill the job, keep it down past the timeout in staging, and verify you get an alert rather than silently missing settlements.
Side effects outside the protocol
Effects outside the protocol need their own design. Move external calls to the edge of the pipeline: emit an intent record to a committed topic, and have a separate consumer perform the call with an idempotency key derived from the record, such as the payment id. The receiving service deduplicates on that key. This is the same reasoning as the transactional outbox, applied to streams.
Upgrades without breaking the chain
Deployments are planned failures, and the chain must survive them too. In Flink, stop the job with a savepoint rather than cancelling it: stop-with-savepoint drains the pipeline, takes a final snapshot and commits the last sink transaction, so the new version resumes from a clean point. Give every stateful operator an explicit uid so its state can be matched to the new job graph after you add or reorder operators; auto-generated ids change with the topology and the state is then dropped or the restore fails.
Keep the transactional id prefix stable across versions of the same job, and different between jobs, so recovery can find and resolve its own open transactions without touching anyone else's. Treat state schema changes like database migrations: use serializers that support evolution, and test restoring a production savepoint into the new version before the release, not during it.
Failure modes
| Failure mode | Symptom | Mitigation |
|---|---|---|
| Sink left at at-least-once | Duplicates downstream after every restart | Assert the delivery guarantee in a startup check |
| Consumers on read_uncommitted | Readers see aborted data | Set isolation.level=read_committed on every downstream consumer |
| Changed transactional id prefix | Lingering open transactions stall read_committed readers | Keep the prefix stable; abort orphans when changing it |
| Non-deterministic processing | Replay produces different output than the first run | Avoid wall-clock time and random ids; derive from event data |
| Checkpoints failing repeatedly | Output stops while input lag grows | Alert on time since last completed checkpoint |
Operating it
Watch time since the last completed checkpoint, checkpoint duration and size, and the lag between processing and committed output, which with a transactional sink is roughly one checkpoint interval. Because consumers on read_committed cannot read past the oldest open transaction on a partition, a stuck transaction shows up as consumer lag downstream, not as an error in the job.
Add an end-to-end audit for the outputs that matter. For a settlement pipeline, a daily batch job can sum the input topic per merchant and compare it with the committed output; any difference is a broken link somewhere in the chain. Such reconciliation catches configuration drift, such as a sink quietly redeployed at a weaker guarantee, that no single metric would reveal.
Trade-offs
Transactional sinks give the strongest guarantee and the highest output latency. Idempotent upserts are cheaper and faster but expose intermediate values during replay. Offsets-in-sink is simple and tight but ties the job to one store. When every consumer is an upsert keyed on a deterministic id, at-least-once delivery into an idempotent sink gives the same final state for less machinery.
What to do next
- Draw your pipeline as source, state, sink and mark which protocol each link uses.
- Verify source retention exceeds your worst outage plus recovery time.
- Pick a sink protocol per output from the table above and assert it at startup.
- Set the transaction timeout above checkpoint interval plus checkpoint duration plus restart time, within the broker maximum.
- Move external calls behind a committed topic and idempotency keys.
- Rehearse the crash points from the worked example in staging, including an outage longer than the timeout.
Keep learning: Kafka exactly-once semantics, the broker side of Kafka transactions, Flink checkpoint design, Flink savepoints, idempotency and the outbox pattern.