Gowtham Potureddi

Posted on Aug 16

Exactly-Once Semantics in Streaming: Idempotency, Transactions & Two-Phase-Commit Sinks

#python #sql #interview #dataengineering

exactly-once semantics is the phrase that ends more streaming design reviews in an argument than any other, because almost everyone repeats the marketing line ("this system gives you exactly-once!") and almost no one can say what actually happens to a single record when a worker crashes halfway through processing it. The honest version is less magical and far more useful: a real streaming system still delivers a record more than once on failure — the network and the two-generals problem guarantee that — but it arranges for the duplicate delivery to have no duplicate effect. That is the whole trick, and it is built from exactly three moving parts that you can learn cold: making writes idempotency-safe, wrapping the read-process-write cycle in a transaction, and coordinating the external sink's commit with the engine's checkpoint through a two-phase commit.

This guide takes those three parts one at a time and shows the real Kafka, Flink, and Spark Structured Streaming code that implements them, so that the next time someone says "just turn on exactly-once," you can point at the exact config flag, the exact commit protocol, and the exact failure it protects against. We start with the three delivery guarantees and why the difference between at-least-once vs exactly-once is really a difference in what the sink does with a redelivered record; then we build up through dedup keys and idempotent producers, Kafka transactions with transactional.id and read_committed, checkpoint-coordinated transactional sinks, and finally the end-to-end chain — replayable source, checkpointed state, transactional sink — that has to hold together for a pipeline to be genuinely exactly-once from ingestion to storage. Every domain pairs a teaching block with a worked interview scenario: the code, the step-by-step trace, the output, and a concept-by-concept breakdown of why it is correct.

When you want hands-on reps alongside the reading, drill the streaming practice library →, rehearse pipeline design on the ETL practice library →, and sharpen event-driven thinking with the event-processing practice library →.

On this page

Delivery semantics — at-most-once, at-least-once, exactly-once
Idempotency & dedup — the cheapest path to effectively-once
Transactions — Kafka EOS, transactional producer/consumer, read_committed
Two-phase-commit / transactional sinks — checkpoint-coordinated commit
End-to-end EOS across source → process → sink; failure & replay
Cheat sheet — exactly-once recipes
Frequently asked questions
Practice on PipeCode

1. Delivery semantics — at-most-once, at-least-once, exactly-once

There are only three guarantees, and they differ in whether a record can be dropped, duplicated, or neither

The one-sentence framing that fixes the whole topic in place: every streaming system offers one of exactly three delivery guarantees — at-most-once (a record may be lost but never duplicated), at-least-once (a record is never lost but may be duplicated), or exactly-once (a record is neither lost nor duplicated in its effect) — and the entire engineering discipline of "exactly-once" is the work of turning an at-least-once transport into an exactly-once effect by deduplicating or transactionally committing the duplicates that the transport will inevitably deliver. You cannot buy exactly-once as a network property; you build it as an application property on top of at-least-once delivery.

The three guarantees, precisely.

At-most-once. The system acknowledges or advances its position before it finishes processing. If it crashes mid-flight, the record is gone and never retried. Zero duplicates, possible data loss. This is what you get from "fire and forget" producers (acks=0) and consumers that commit offsets before processing.
At-least-once. The system finishes processing (or attempts to) and only then advances its position, and it retries on any uncertainty. If it crashes after processing but before recording that it processed, it will reprocess on restart. Zero data loss, possible duplicates. This is the sane default for most transports, including Kafka's out-of-the-box behaviour.
Exactly-once (effectively-once). The system delivers at-least-once and makes the duplicate harmless — either the write is idempotent (a replay overwrites the same row) or the write and the position-advance commit atomically in one transaction (a replay is rolled back). Zero loss, zero net duplicates.

Why "exactly-once" is a lie about the transport and the truth about the effect. People new to the topic imagine a magic wire that delivers each byte precisely once. That wire cannot exist: the sender must retry when an acknowledgement is lost, and it cannot tell "the message was lost" from "the ack was lost." The two-generals problem is a theorem, not a bug. So a message will be sent twice sometimes. Exactly-once systems accept that and guarantee the observable outcome is as if each message were processed once. That is why practitioners increasingly say effectively-once — it is a more honest name for the same guarantee.

The end-to-end chain is only as strong as its weakest link. A pipeline is source → process → sink, and exactly-once must hold across all three:

Source must be replayable. On restart the system must be able to re-read from a known position (Kafka offsets, a file byte-offset, a WAL LSN). A non-replayable source (a UDP feed, a fire-and-forget socket) caps you at at-most-once — you cannot re-fetch what you dropped.
Processing state must be recoverable atomically with the source position. If the operator holds state (a running count, a windowed aggregate), that state and the source offset must be snapshotted together so that on recovery the position and the state agree. This is what checkpointing does.
Sink must be idempotent or transactional. The replayed writes that recovery produces must not create duplicate rows. Either the sink dedups them (idempotent) or the engine commits them transactionally in lockstep with the checkpoint (two-phase commit).

Break any one link and the whole pipeline degrades to the weakest guarantee. A perfectly transactional Flink job that writes to a sink which appends blindly is at-least-once at the sink, and therefore at-least-once end-to-end.

Why at-most-once is almost never what you want, and when it is. At-most-once is tempting because it is the simplest and lowest-latency (no retries, no dedup bookkeeping). But losing data silently is unacceptable for anything financial, transactional, or audited. The narrow cases where at-most-once is fine: high-volume telemetry where a dropped sample changes nothing (approximate dashboards, sampled metrics), or where a newer value fully supersedes an older one (a "latest sensor reading" gauge). Everywhere else, start at at-least-once and add idempotency or transactions to reach exactly-once.

Worked example — tracing one record through a crash under each semantic

Detailed explanation. The clearest way to internalise the three guarantees is to follow a single record — call it order-42 — through a consumer that reads it, processes it (charges a card), and records progress, then crash the consumer at the worst possible moment and see what each semantic does. The "worst moment" is after the side effect but before the position is durably recorded, because that is exactly the window the two-generals problem opens. The order in which you (a) advance the offset and (b) perform the side effect is what determines which guarantee you get.

Offset-first (commit before processing) → at-most-once. A crash after the commit but before the charge means the charge never happens and is never retried.
Process-first (commit after processing) → at-least-once. A crash after the charge but before the commit means the charge is retried on restart → double charge unless the charge is idempotent.
Process-first + idempotent/transactional effect → exactly-once. The retried charge is deduplicated (same order_id already charged) or rolled back, so the net effect is one charge.

Question. A consumer reads order-42, charges a card, and commits the offset. If it crashes between two of those steps, which ordering yields at-most-once, at-least-once, and exactly-once?

Input.

Ordering of steps	Crash point	Outcome for `order-42`
commit offset → charge	after commit, before charge	never charged (lost)
charge → commit offset	after charge, before commit	charged twice on retry
charge (idempotent) → commit offset	after charge, before commit	charged once (retry deduped)

Code.

# The three orderings as pseudocode
# (A) AT-MOST-ONCE: advance position first
commit_offset(order.offset)      # <- crash here => charge never runs
charge_card(order)

# (B) AT-LEAST-ONCE: do the work first, then advance
charge_card(order)               # <- crash here => on restart we re-read and charge AGAIN
commit_offset(order.offset)

# (C) EXACTLY-ONCE: idempotent work, then advance
if not already_charged(order.id):    # dedup key = order_id
    charge_card(order)               # <- crash here => restart re-reads, sees already_charged, skips
commit_offset(order.offset)

Step-by-step trace.

In (A), the offset is committed first; a crash before charge_card means restart resumes after order-42, so it is never charged — the record is silently lost (at-most-once).
In (B), the charge runs, then the crash prevents the commit; on restart the consumer re-reads order-42 (offset never advanced) and charges again — a duplicate (at-least-once).
In (C), the charge is guarded by an idempotency check keyed on order_id; on restart the re-read re-runs the guard, already_charged(order-42) is true, the charge is skipped, and the offset finally commits — one net charge (exactly-once).
The transport in (B) and (C) is identical at-least-once delivery; only the effect differs, which is the entire point of "effectively-once."

Output:

Semantic	Duplicates?	Data loss?	How to get it
At-most-once	no	yes	commit position before doing the work
At-least-once	yes	no	do the work, then commit position
Exactly-once	no	no	at-least-once + idempotent or transactional effect

Rule of thumb. If a system claims exactly-once, ask two questions: "what makes the source replayable?" and "what makes the sink write idempotent or transactional?" — if it cannot answer both, it is at-least-once wearing a marketing badge.

2. Idempotency & dedup — the cheapest path to effectively-once

Make the write harmless to repeat, and at-least-once delivery becomes exactly-once for free

The invariant to burn in: an operation is idempotent when applying it twice has the same effect as applying it once — f(f(x)) = f(x) — so if every write your pipeline makes is idempotent, then at-least-once delivery is already exactly-once, because the duplicate writes that failure produces simply overwrite themselves instead of accumulating. Idempotency is the cheapest exactly-once mechanism because it needs no distributed transaction, no coordinator, and no two-phase protocol — just a stable key and a write that is safe to repeat.

The building blocks of an idempotent sink.

A dedup key (business key). Every record needs a stable identity — event_id, (user_id, event_ts), an upstream primary key — that is the same on the original and on the replay. Without a stable key you cannot tell a replay from a genuinely new record.
An idempotent write, not an append. INSERT accumulates duplicates; UPSERT / MERGE / INSERT … ON CONFLICT DO UPDATE collapse a repeat into an overwrite of the same row. Writing to a keyed store (a KV by primary key, an object at a deterministic path) is naturally idempotent.
A dedupe window when you cannot upsert. If the sink can only append (a log, a raw event table), keep a bounded cache of recently-seen keys — often in the stream engine's keyed state with a TTL — and drop a record whose key you have already emitted within the window.
Idempotency at the producer. Kafka's idempotent producer stamps each record with a producer id (PID) and a per-partition sequence number so the broker can drop a retried duplicate before it ever hits the log.

Dedup keys — where the identity comes from. The most common interview mistake is deduping on the wrong thing. Dedup on a content hash and two legitimately identical events (two clicks on the same button in the same millisecond) collapse into one — data loss. Dedup on an arrival timestamp and a replay of the same event looks new — a duplicate. The right key is the record's business identity: an event_id minted at the source, or a natural composite key that uniquely names the real-world event. If the source does not mint an id, add one at the earliest possible point (ideally the producer) so it is stable across every retry.

Dedupe windows and state TTL — bounding the memory. You cannot remember every key you have ever seen; the seen-keys set would grow without bound. In practice you dedup within a window — "drop a key I have seen in the last N hours" — because the failure-and-replay gap is bounded (a checkpoint interval, a retry timeout), not infinite. Stream engines make this easy: hold the seen-keys in keyed state with a TTL, and the engine evicts old keys automatically. The window must be at least as long as your worst-case redelivery delay, or a late replay slips through as a duplicate.

Idempotent producer — dedup on the broker. Kafka's idempotent producer (enabled with enable.idempotence=true, the default since Kafka 3.0) solves the producer-retry duplicate specifically. Without it, a producer that sends a batch, has the ack lost, and retries writes the batch twice. With it, the broker tracks the highest sequence number it has committed per (producer_id, partition) and silently drops any record whose sequence number it has already seen — so a retry is deduplicated at the log, not at your sink. It guarantees exactly-once per producer session, per partition; it does not by itself dedup across producer restarts or across partitions (that is what transactions add).

Common traps to pre-empt.

Deduping on a non-stable key (content hash, arrival time) — collapses real duplicates or misses replays. Use a business event_id.
INSERT instead of UPSERT into the sink — every replay appends a new row. Use MERGE/ON CONFLICT.
An unbounded dedupe set — memory grows forever. Bound it with a windowed TTL sized to your redelivery gap.
Assuming the idempotent producer gives end-to-end EOS — it only removes producer-retry duplicates on one partition/session; the consumer side and the sink still need their own idempotency or a transaction.

Idempotent upsert sink keyed on event_id — a worked teaching example

Detailed explanation. The most common effectively-once sink is an upsert keyed on the record's business id. An at-least-once stream may deliver event_id = e-42 twice (once originally, once after a replay), and the sink must land exactly one row for it. You achieve this with MERGE (or INSERT … ON CONFLICT DO UPDATE): match on the dedup key, update if present, insert if absent. Because a second application matches the row it wrote the first time and overwrites it with identical values, the operation is idempotent — the table looks the same whether the event arrived once or five times.

Match on the dedup key (event_id), never on all columns.
Update-or-insert so a replay overwrites rather than appends.
Deterministic payload — the same event must map to the same row values so the overwrite is a no-op-in-effect.
A unique constraint on the dedup key backs the upsert and catches any bug that tries to double-insert.

Question. An at-least-once stream may redeliver an event; write it to a warehouse table so each event_id lands exactly one row regardless of how many times it is delivered.

Input.

Field	Type	Role
`event_id`	STRING	dedup key (unique)
`user_id`	STRING	payload
`amount`	NUMERIC	payload
`event_ts`	TIMESTAMP	payload

Code.

-- Target table with a UNIQUE dedup key so duplicates cannot accumulate
CREATE TABLE fact_payments (
  event_id  STRING     NOT NULL,
  user_id   STRING,
  amount    NUMERIC,
  event_ts  TIMESTAMP,
  PRIMARY KEY (event_id)      -- the dedup key
);

-- Idempotent write: MERGE the micro-batch on event_id (upsert)
MERGE INTO fact_payments AS t
USING staged_batch AS s
ON t.event_id = s.event_id                 -- match on the dedup key ONLY
WHEN MATCHED THEN UPDATE SET
  user_id  = s.user_id,
  amount   = s.amount,
  event_ts = s.event_ts
WHEN NOT MATCHED THEN INSERT
  (event_id, user_id, amount, event_ts)
  VALUES (s.event_id, s.user_id, s.amount, s.event_ts);

-- Postgres equivalent for a row-at-a-time sink:
-- INSERT INTO fact_payments (event_id,user_id,amount,event_ts)
-- VALUES (:id,:uid,:amt,:ts)
-- ON CONFLICT (event_id) DO UPDATE SET
--   user_id=EXCLUDED.user_id, amount=EXCLUDED.amount, event_ts=EXCLUDED.event_ts;

Step-by-step trace.

The first delivery of e-42 finds no matching row → WHEN NOT MATCHED inserts it → one row.
Processing crashes downstream; on recovery the stream replays and delivers e-42 again in a later batch.
The replay now finds the existing row → WHEN MATCHED updates it with identical values → still one row, unchanged in effect.
The PRIMARY KEY (event_id) guarantees the table can physically hold only one row per id, so even a buggy blind insert would be rejected rather than duplicated.

Output:

Delivery	Rows for `e-42` after write
first (insert)	1
replay (matched update)	1 (overwritten, not added)
third replay	1 (still idempotent)

Rule of thumb. If the sink supports a unique key and an upsert, idempotency is the whole exactly-once story — you do not need transactions; you need MERGE on a stable business key.

Kafka idempotent producer — sequence numbers reject the retry — a worked teaching example

Detailed explanation. Before you reach for a full transaction, know what the idempotent producer alone buys you: it removes the producer-retry duplicate. When enable.idempotence=true, the producer obtains a producer id (PID) and tags each record on each partition with a monotonically increasing sequence number. The broker remembers the last sequence number it committed per (PID, partition); if a retried batch arrives with a sequence number it has already seen, it acknowledges but does not append — the duplicate is dropped at the log. Turning it on also forces the safe settings that make it work: acks=all, bounded in-flight requests, and infinite-ish retries.

PID + sequence number = the broker's dedup key for producer retries.
acks=all so a record is only "committed" once all in-sync replicas have it.
max.in.flight.requests.per.connection ≤ 5 so retries cannot reorder past the dedup window.
Scope: one producer session, per partition — not across restarts, not across partitions (transactions extend it there).

Question. A producer sends a batch, the ack is lost, and the client retries the same batch. Configure Kafka so the retry does not create a duplicate on the topic.

Input.

Setting	Value	Purpose
`enable.idempotence`	`true`	PID + per-partition sequence numbers
`acks`	`all`	commit only after all ISR replicas ack
`retries`	high / `Integer.MAX_VALUE`	keep retrying the ambiguous send
`max.in.flight.requests.per.connection`	`5` (≤5)	preserve ordering under retry

Code.

// Idempotent producer: retries can no longer duplicate on a partition
Properties p = new Properties();
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");   // PID + seq numbers
p.put(ProducerConfig.ACKS_CONFIG, "all");                  // implied by idempotence
p.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);   // safe: dedup on the broker
p.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

try (KafkaProducer<String,String> producer = new KafkaProducer<>(p)) {
    // seq numbers 0,1,2… are attached automatically per (PID, partition)
    producer.send(new ProducerRecord<>("payments", "acct-7", "charge:42"));
    // If the ack is lost and the client retries seq=N, the broker sees it
    // already committed seq=N and drops the retry — no duplicate on the log.
}

Step-by-step trace.

The producer gets a PID and sends record with sequence number N on partition p; the broker appends it and remembers "last committed seq for (PID, p) = N."
The broker's ack is lost on the network; the client cannot tell whether the append happened, so it retries the same record (still sequence N).
The broker receives sequence N again, compares against its stored high-water sequence, sees N is already committed, and drops the retry while still acking the client.
The topic contains exactly one copy; the producer-retry duplicate is eliminated at the broker without any consumer-side dedup.

Output:

Event	Broker action	Copies on topic
send seq=N	append, record high seq=N	1
retry seq=N (ack lost)	recognise duplicate, drop, re-ack	1
send seq=N+1	append	2 (distinct record)

Rule of thumb. Turn enable.idempotence=true on by default for every producer — it is nearly free and removes the single most common source of duplicate messages (producer retries); reach for transactions only when you also need atomic multi-partition writes or read-process-write.

Interview scenario on deduplicating an at-least-once stream

You consume an at-least-once Kafka topic of user events and must load them into a warehouse dim_events table so analysts never see a duplicate, even though the pipeline redelivers events after every restart. Events carry a source-minted event_id. Design the sink write.

Solution Using a MERGE upsert on the event_id dedup key

Answer choices (as an interviewer would present them).

A. INSERT every consumed event into dim_events and run a nightly SELECT DISTINCT cleanup job.
B. Stage each micro-batch and MERGE into dim_events on event_id (upsert), with a unique constraint on event_id.
C. Dedup in the consumer by hashing the full event payload and skipping repeats seen this run.
D. Write to a new partition every run and let analysts pick the latest.

Code.

Elimination:
A  INSERT + nightly DISTINCT -> duplicates visible all day, cleanup is O(table)   [reject: not exactly-once]
C  hash-of-payload dedup     -> collapses legitimately-identical events; in-memory
                               set lost on restart -> replay duplicates            [reject: wrong key + not durable]
D  new partition per run     -> unbounded partitions, analysts see dupes across    [reject: doesn't dedup]
B  MERGE on event_id + unique constraint -> idempotent upsert, durable, exact      [ACCEPT]

Step-by-step trace.

Constraints: at-least-once source (replays after restart), analysts must never see a duplicate, a stable event_id exists.
A leaves duplicates visible until a nightly job and scales badly — it does not make the write idempotent, so it fails the "never see a duplicate" bar — eliminate.
C dedups on the wrong key (payload hash collapses two real identical events) and keeps the seen-set in memory, so a restart forgets it and the replay duplicates — eliminate.
D never actually deduplicates; it multiplies partitions and pushes the problem onto analysts — eliminate.
B stages the batch and MERGEs on event_id: the first delivery inserts, every replay updates the same row, and the unique constraint guarantees one physical row per id — idempotent and durable.

Output:

Requirement	Mechanism
Never show a duplicate	`MERGE` upsert on `event_id`
Survive restarts/replays	durable table + unique key (not in-memory)
Correct identity	source-minted `event_id`, not a payload hash
Bounded cost	O(batch) upsert, no full-table cleanup

Why this works — concept by concept:

Stable dedup key — matching on the source-minted event_id is what lets the sink recognise a replay as the same event; a payload hash or timestamp would either collapse real duplicates or miss replays.
Idempotent upsert — MERGE/ON CONFLICT turns a repeated delivery into an overwrite of the identical row, so the table's state is a function of the set of events, not the count of deliveries.
Durable, not in-memory, dedup — the guarantee lives in the table's unique constraint, which survives restarts, unlike an in-process seen-set that a crash forgets.
Cost — the write is O(batch size) with an index lookup per key (roughly O(log n) per row); there is no nightly O(table) dedup pass, so cost scales with new data, not accumulated data.

Streaming
Topic — streaming
Streaming dedup and idempotent-sink problems

Practice →

Event processing Topic — event-processing Event identity and deduplication problems

Practice →

3. Transactions — Kafka EOS, transactional producer/consumer, read_committed

When the write must be atomic with the offset, wrap read-process-write in one transaction

The invariant: when a stream processor reads from Kafka, transforms, and writes back to Kafka, exactly-once requires that the output records and the advance of the input offsets commit atomically — all-or-nothing — and Kafka transactions provide exactly that by binding the produced messages and the consumed offsets into one transaction that a downstream read_committed consumer will only ever see if it commits. This is the "consume-transform-produce" or read-process-write pattern, and it is the backbone of kafka exactly once for stream-to-stream pipelines.

The pieces of Kafka EOS (exactly-once semantics).

transactional.id. A stable, unique id per logical producer instance. It ties a producer across restarts to the same transaction and enables zombie fencing — when a new instance with the same transactional.id calls initTransactions, the coordinator bumps an epoch and fences out the old (zombie) instance so it can no longer commit.
initTransactions / beginTransaction / commitTransaction / abortTransaction. The lifecycle. initTransactions registers the producer and recovers any dangling transaction; beginTransaction opens one; commitTransaction atomically finalises all produced records and offsets; abortTransaction discards them.
sendOffsetsToTransaction. The linchpin: it adds the consumed input offsets to the same transaction as the produced output. Committing the transaction advances the input offset and publishes the output as one atomic act — so you cannot advance past an input without its output being durable, and vice versa.
isolation.level=read_committed. Downstream consumers with this setting never read records from open or aborted transactions; they only see committed data, up to the last stable offset (LSO). With the default read_uncommitted, a consumer would see aborted records and the guarantee evaporates.

The transaction coordinator and the internal two-phase commit. Kafka implements transactions with a broker-side transaction coordinator and an internal transaction-state log (an internal topic). On commit, the coordinator runs a two-phase protocol: it writes prepare-commit markers, then writes commit markers into each affected partition (and the consumer-offsets partition), and finally records the transaction as complete. A read_committed consumer uses those markers to decide what is visible; anything past the LSO (belonging to an in-flight transaction) is withheld. This is a genuine two-phase commit inside Kafka — which is why the next section's external-sink two-phase commit feels familiar.

Atomic read-process-write — the exactly-once loop. The pattern that gives you EOS between two Kafka topics: assign the consumer, read a batch, produce the transformed output, call sendOffsetsToTransaction to fold the input offsets into the transaction, and commitTransaction. If anything fails, abortTransaction throws away both the output and the offset advance, and the next run re-reads the same input — no partial effect ever escapes.

Common traps to pre-empt.

Committing offsets with consumer.commitSync() instead of sendOffsetsToTransaction — that decouples the offset from the output, breaking atomicity. Offsets must ride inside the producer transaction.
Downstream consumer left at read_uncommitted — it will read aborted/uncommitted records; you must set read_committed end-to-end or the producer-side transaction is pointless.
Reusing a random transactional.id per run — you lose zombie fencing and recovery of dangling transactions; the id must be stable per logical instance.
Expecting exactly-once to a non-Kafka sink from Kafka transactions alone — Kafka transactions only cover Kafka topics and the offsets topic; an external DB/file sink needs the two-phase-commit sink of the next section.

The exactly-once consume-transform-produce loop — a worked teaching example

Detailed explanation. This is the canonical Kafka EOS pattern and a frequent interview ask: build a processor that reads topic A, transforms, writes topic B, and is exactly-once even across crashes and consumer-group rebalances. The trick is that the output to B and the committed offset on A are the same transaction, so a crash can never leave "output written but offset not advanced" (a duplicate) or "offset advanced but output not written" (a loss). The consumer that reads A must have enable.auto.commit=false because the producer, not the consumer, commits the offsets — transactionally.

Producer has a stable transactional.id and calls initTransactions once.
Consumer has enable.auto.commit=false and (if it also reads transactional input) isolation.level=read_committed.
Each cycle: beginTransaction → produce outputs → sendOffsetsToTransaction(inputOffsets, groupMetadata) → commitTransaction.
On failure: abortTransaction and re-poll; the input offsets were never advanced.

Question. Read from topic A, transform, and write to topic B with exactly-once semantics across crashes and rebalances. Write the loop.

Input.

Component	Setting
Producer	`transactional.id = "etl-a-to-b-0"`, `enable.idempotence=true`
Consumer	`enable.auto.commit=false`, `isolation.level=read_committed`, `group.id="etl"`
Atomic unit	produced records to `B` + consumed offsets on `A`

Code.

producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "etl-a-to-b-0");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

KafkaProducer<String,String> producer = new KafkaProducer<>(producerProps);
KafkaConsumer<String,String> consumer = new KafkaConsumer<>(consumerProps);
producer.initTransactions();               // register txn.id, fence zombies, recover danglers
consumer.subscribe(List.of("A"));

while (running) {
    ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(200));
    if (records.isEmpty()) continue;

    producer.beginTransaction();
    try {
        for (ConsumerRecord<String,String> r : records) {
            String out = transform(r.value());
            producer.send(new ProducerRecord<>("B", r.key(), out));   // output
        }
        // Fold the INPUT offsets into the SAME transaction:
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        for (TopicPartition tp : records.partitions()) {
            long last = records.records(tp).get(records.records(tp).size()-1).offset();
            offsets.put(tp, new OffsetAndMetadata(last + 1));
        }
        producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
        producer.commitTransaction();       // output to B + offsets on A commit atomically
    } catch (KafkaException e) {
        producer.abortTransaction();         // discard output AND offset advance; re-poll
    }
}

Step-by-step trace.

initTransactions() registers etl-a-to-b-0, bumps its epoch (fencing any older zombie instance), and rolls back any transaction the previous instance left dangling.
The loop polls a batch from A, opens a transaction, and produces the transformed records to B — none of them are yet visible to a read_committed consumer.
sendOffsetsToTransaction attaches the next offsets for A to the open transaction, so the offset advance is now part of the same atomic unit as the output.
commitTransaction writes commit markers to B's partitions and to the consumer-offsets partition together; only now does topic B's output become visible and the group's offset on A advance.
If the worker crashes before the commit, the transaction aborts on recovery: B never shows the output and A's offset never advanced, so the next run re-reads and reprocesses the same batch — exactly-once.

Output:

Failure point	Output on B	Offset on A	Net effect
crash before commit	rolled back (invisible)	not advanced	batch reprocessed cleanly
commit succeeds	visible once	advanced once	processed exactly once
zombie old instance tries to commit	fenced (rejected)	unchanged	no double write

Rule of thumb. For Kafka-to-Kafka exactly-once, the offsets must ride inside the producer transaction via sendOffsetsToTransaction, and every downstream consumer must be read_committed — those two facts are the whole pattern.

Turning on framework-level EOS — Kafka Streams and Flink config — a worked teaching example

Detailed explanation. You rarely hand-write the transaction loop in production; you let a framework do it and just flip a flag. Kafka Streams exposes it as processing.guarantee=exactly_once_v2, which manages the transactional.id, the offset-in-transaction commit, and the state-store changelog atomicity for you. Flink expresses the same idea through checkpointing plus an exactly-once Kafka sink (DeliveryGuarantee.EXACTLY_ONCE) that uses Kafka transactions committed on checkpoint completion. Knowing which flag and what it manages under the hood is the interview signal — the flag is not magic, it is the loop above, automated.

Kafka Streams: processing.guarantee=exactly_once_v2 — one producer per instance, offsets and state-changelog folded into the transaction, committed on the commit interval.
Flink: env.enableCheckpointing(...) + a Kafka sink set to EXACTLY_ONCE with a transactionalIdPrefix — the sink pre-commits on snapshot and commits on checkpoint-complete.
exactly_once_v2 improved on the original by using a single producer per instance (fewer resources) and requiring brokers ≥ 2.5.
The commit interval is a latency knob — output only becomes visible to read_committed consumers at each transaction commit.

Question. Configure Kafka Streams and Flink for exactly-once without hand-writing the transaction loop.

Input.

Framework	Key setting	What it manages
Kafka Streams	`processing.guarantee=exactly_once_v2`	txn id, offsets, state changelog
Flink	checkpointing + Kafka sink `EXACTLY_ONCE`	pre-commit/commit on checkpoint
Both	downstream `isolation.level=read_committed`	hide aborted output

Code.

// --- Kafka Streams: one flag turns on EOS ---
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-agg");   // becomes the txn.id base
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);        // txn commit cadence (latency knob)
// state stores, offsets, and output all commit in one transaction per interval

// --- Flink: checkpointing + exactly-once Kafka sink ---
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(30_000);   // snapshot state + offsets every 30s
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("broker:9092")
    .setRecordSerializer(/* ... topic "B" ... */ serializer)
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)      // Kafka transactions
    .setTransactionalIdPrefix("flink-a-to-b")                  // stable prefix for fencing
    .build();

Step-by-step trace.

In Kafka Streams, EXACTLY_ONCE_V2 makes the runtime open a transaction per commit interval, fold the input offsets and the state-store changelog records into it, and commit them together — the hand-written loop, automated.
COMMIT_INTERVAL_MS=100 sets how often that transaction commits; smaller means lower end-to-end latency (output visible sooner) at the cost of more, smaller transactions.
In Flink, enableCheckpointing snapshots operator state and Kafka source offsets together; the EXACTLY_ONCE Kafka sink opens a Kafka transaction and pre-commits on each snapshot.
When the checkpoint completes globally, Flink tells the sink to commit the Kafka transaction — so output becomes visible only for records whose state and offsets are durably checkpointed.
Downstream, read_committed consumers see topic B's records only after these commits, so a failed checkpoint's output is aborted and never observed.

Output:

Framework	Flag	Visible to read_committed when
Kafka Streams	`exactly_once_v2`	each commit-interval transaction commits
Flink	checkpointing + `EXACTLY_ONCE` sink	the enclosing checkpoint completes
Either at `read_uncommitted`	—	immediately (guarantee lost)

Rule of thumb. Prefer the framework flag (exactly_once_v2 / Flink EXACTLY_ONCE sink) over a hand-rolled loop, and remember the commit interval / checkpoint interval is your latency-vs-overhead dial — smaller commits, lower latency, more transaction overhead.

Interview scenario on a money-movement stream

A payments service consumes a transfers topic, applies a fee transformation, and produces to a ledger topic. During a consumer-group rebalance last week, some transfers were produced to ledger twice, double-crediting accounts. Redesign so that a rebalance or crash can never double-produce.

Solution Using a transactional read-process-write with sendOffsetsToTransaction

Answer choices.

A. Keep at-least-once producing but add a nightly reconciliation job that reverses duplicate ledger entries.
B. Wrap consume-transform-produce in a Kafka transaction with a stable transactional.id, fold offsets in via sendOffsetsToTransaction, and set downstream isolation.level=read_committed.
C. Turn on enable.idempotence=true on the producer and consider it solved.
D. Have the consumer commitSync() offsets immediately after producing, before the batch is fully acked.

Code.

Elimination:
A  nightly reversal      -> duplicates are visible and acted on for hours; reconciliation
                            is compensating, not exactly-once                        [reject]
C  idempotence only       -> removes producer-RETRY dupes on a partition, but NOT the
                            rebalance case where a re-read reprocesses and re-produces  [reject: wrong scope]
D  commitSync after produce -> offset + output not atomic; crash between them duplicates
                              or loses; also races the rebalance                       [reject]
B  txn + sendOffsetsToTransaction + read_committed -> atomic output+offset, fenced      [ACCEPT]

Step-by-step trace.

Constraints: a rebalance/crash must never double-produce to ledger; the fix must be exactly-once, not compensating.
A tolerates the duplicate and reverses it later — accounts are wrong in the meantime and it is a reconciliation hack, not a guarantee — eliminate.
C only removes producer-retry duplicates on one partition/session; the actual bug is a reprocessing duplicate after rebalance (the batch is re-read and re-produced), which idempotence does not cover — eliminate.
D commits the offset separately from the produce, so a crash in between still duplicates or loses, and it races the rebalance — eliminate.
B binds the produced ledger records and the advanced transfers offsets into one transaction; the stable transactional.id fences the pre-rebalance zombie so it cannot commit, and read_committed hides any aborted attempt — a rebalance can no longer double-credit.

Output:

Failure	Under at-least-once (bug)	Under transactional EOS
rebalance mid-batch	batch re-produced → double credit	zombie fenced, batch reprocessed once
crash after produce	offset not committed → dup	output aborted, offset not advanced
downstream read	sees dup	`read_committed` hides aborted

Why this works — concept by concept:

Atomic output-plus-offset — sendOffsetsToTransaction puts the ledger write and the transfers offset advance in one commit, so "produced but not advanced" (the duplicate) is structurally impossible.
Zombie fencing via transactional.id — the epoch bump on initTransactions stops a stalled pre-rebalance instance from committing a stale transaction, which is exactly the rebalance double-produce that hurt the ledger.
read_committed isolation — downstream consumers only observe committed transactions, so even an aborted attempt is never seen as a credit.
Cost — transactions add coordinator round-trips and hold output invisible until commit, so end-to-end latency rises by roughly the commit interval and throughput drops modestly — the price of correctness on money movement.

Streaming
Topic — streaming
Transactional stream-processing problems

Practice →

ETL Topic — etl Read-process-write pipeline problems

Practice →

4. Two-phase-commit / transactional sinks — checkpoint-coordinated commit

When the sink lives outside the engine, commit it in two phases tied to the checkpoint

The invariant: Kafka transactions cover Kafka, but the moment your sink is an external system — a database, a file system, an object store — its write is not part of the engine's checkpoint, so exactly-once requires a two-phase commit that stages the write during the checkpoint (pre-commit) and finalises it only once the checkpoint is globally complete (commit), with the commit made idempotent so recovery can safely re-run it. This is the pattern behind Flink's TwoPhaseCommitSinkFunction and Spark Structured Streaming's idempotent sinks.

Why the naive sink breaks. A streaming engine achieves fault tolerance by periodically checkpointing (snapshotting) its state and source offsets. On recovery it rewinds to the last checkpoint and replays everything after it. If the sink wrote directly and eagerly, those replayed records would be written again — duplicates. The external sink is not covered by the checkpoint, so you must coordinate its commit with the checkpoint lifecycle. That coordination is a two-phase commit, with the engine's checkpoint coordinator acting as the transaction coordinator.

The two-phase-commit sink protocol (Flink's TwoPhaseCommitSinkFunction). The sink participates in checkpoints through four callbacks:

beginTransaction — start a new external transaction (open a DB transaction, create a temp file, open a Kafka transaction). New records write into it.
preCommit — called when the operator snapshots for a checkpoint. Flush and stage everything so far, then stop writing to this transaction. The data is durable but not yet visible. This is phase one ("prepared").
commit — called after the entire checkpoint completes (a global signal, via notifyCheckpointComplete). Finalise the staged transaction so its data becomes visible. This is phase two.
abort — if the checkpoint fails or the job restarts before commit, discard the staged transaction.

Why the pre-commit / commit split gives exactly-once. The checkpoint barrier is the synchronisation point. A record's effect is only committed if its checkpoint completed; if the job crashes, it restores from the last completed checkpoint, whose transaction was already committed, and replays only records after it — into a fresh transaction. The subtle-but-critical requirement: commit must be idempotent, because a crash can happen after a checkpoint completes but before the commit callback finishes, so recovery must re-issue the commit and it must be safe to commit the same (already-committed) transaction twice. Staging to a temp file and committing with an atomic rename is idempotent (renaming to an existing target, or re-running a rename that already happened, lands the same file once). Committing a Kafka transaction by id is idempotent for the same reason.

Spark Structured Streaming's take — batchId idempotence. Spark's micro-batch model assigns every batch a monotonic batchId. Its built-in file sink writes each batch atomically and records committed batches in a _spark_metadata log, so a re-run of batchId=128 is skipped — the file sink is exactly-once. For arbitrary sinks you use foreachBatch(fn) where fn receives (microBatchDF, batchId); you make the write idempotent by keying on batchId (e.g. MERGE guarded by batch id, or a Delta transaction that ignores an already-committed batch), so a replayed batch is a no-op. Delta Lake's txnAppId/txnVersion do exactly this for you.

Sink idempotence is the common thread. Whether it is Flink's atomic-rename commit, Kafka's commit-by-transaction-id, or Spark's skip-on-batchId, the deep requirement is identical: the final commit step must be idempotent so that a recovery that re-runs it does not duplicate. Two-phase commit gets you atomicity with the checkpoint; idempotent commit gets you safety on replay — you need both.

Common traps to pre-empt.

Committing eagerly in preCommit — data becomes visible before the checkpoint completes, so a failed checkpoint leaks uncommitted records. Commit only in commit.
A non-idempotent commit — a crash between checkpoint-complete and commit re-runs the commit and duplicates. Use atomic rename / commit-by-id / skip-on-batchId.
Forgetting transaction-timeout > checkpoint-interval — for Kafka 2PC sinks, if the transaction times out before the next checkpoint commits, staged data is lost; size transaction.timeout.ms above the checkpoint interval.
Using foreach (per-row) and hoping for exactly-once — per-row sinks have no batch boundary to key idempotence on; use foreachBatch with batchId.

A Flink two-phase-commit file sink — stage then atomically rename — a worked teaching example

Detailed explanation. Consider writing a Flink stream to a filesystem/object store exactly-once. You cannot make an object store transactional, but you can make the commit idempotent with a staged-temp-file-then-atomic-rename pattern coordinated by the checkpoint. During processing you write records to a per-checkpoint temp file; on preCommit (snapshot) you flush and close it (staged, invisible); on commit (checkpoint complete) you atomically rename it to its final visible name. If recovery re-issues the commit, renaming a temp file that was already renamed is a no-op — idempotent.

beginTransaction returns a fresh temp path for this checkpoint.
invoke appends records to the temp file.
preCommit flushes and closes the temp file (durable, not visible).
commit atomically renames temp → final (idempotent); abort deletes the temp file.

Question. Write a Flink sink to an object store so that a crash-and-restore never leaves a partial or duplicated output file.

Input.

Checkpoint phase	Sink action	Visibility
beginTransaction	create `part-<ckpt>.tmp`	none
invoke (per record)	append to temp	none
preCommit (snapshot)	flush + close temp	staged
commit (ckpt complete)	rename temp → `part-<ckpt>`	visible

Code.

// Sketch of a TwoPhaseCommitSinkFunction<IN, TXN=TempFile, CTX=Void>
public class FileTwoPhaseCommitSink
        extends TwoPhaseCommitSinkFunction<Row, TempFile, Void> {

    @Override
    protected TempFile beginTransaction() {                  // phase-0: open a staged txn
        return new TempFile("/out/part-" + UUID.randomUUID() + ".tmp");
    }

    @Override
    protected void invoke(TempFile txn, Row row, Context c) {
        txn.append(serialize(row));                          // write into the staged file
    }

    @Override
    protected void preCommit(TempFile txn) throws IOException {
        txn.flushAndClose();                                 // phase-1: durable but INVISIBLE
    }

    @Override
    protected void commit(TempFile txn) {                    // phase-2: after checkpoint completes
        Path finalPath = txn.finalName();                    // e.g. /out/part-000128
        if (!fs.exists(finalPath)) {                         // idempotent guard
            fs.rename(txn.path(), finalPath);                // atomic rename => visible ONCE
        }                                                    // re-run after recovery = no-op
    }

    @Override
    protected void abort(TempFile txn) {
        fs.deleteIfExists(txn.path());                       // discard staged data on failure
    }
}

Step-by-step trace.

Between checkpoints, records append to part-<ckpt>.tmp; nothing is visible to a downstream reader.
When the checkpoint barrier arrives, preCommit flushes and closes the temp file — it is now durable but still invisible (phase one, "prepared").
Once all operators confirm the checkpoint, Flink calls commit, which atomically renames the temp file to part-000128 — the output appears exactly once (phase two).
If the job crashes after checkpoint 128 completes but before commit returns, recovery re-issues commit; the fs.exists(finalPath) guard (or the fact the rename already happened) makes it a no-op — no duplicate file.
Records written after checkpoint 128 were in a later temp file that never committed; recovery replays them into a fresh transaction, so nothing is lost or doubled.

Output:

Scenario	Files on the store
clean run	one `part-000128`, no temp
crash before commit	staged temp aborted; batch replayed into next txn
crash after commit signal	commit re-run → still one `part-000128`

Rule of thumb. For an external sink, stage during preCommit, make the final commit idempotent (atomic rename / commit-by-id), and never make data visible before the checkpoint completes — that is two-phase commit for streaming in one sentence.

Spark Structured Streaming idempotent sink on batchId — a worked teaching example

Detailed explanation. Spark's micro-batch engine gives you a monotonically increasing batchId for every batch, and that id is the hook for exactly-once to an arbitrary sink. In foreachBatch(fn), fn(microBatchDF, batchId) may be called more than once for the same batchId on failure/restart, so you must make the write idempotent in batchId: either MERGE on a business key (idempotent regardless of batch), or record the highest committed batchId and skip a batch you have already applied. Delta Lake automates the latter with txnAppId + txnVersion, which cause Delta to ignore a write whose (appId, version) it has already committed.

batchId is deterministic and monotonic — the same records always carry the same id on replay.
Idempotent write = MERGE on a business key, or "skip if batchId ≤ last_committed."
Delta txnAppId/txnVersion makes the skip automatic and atomic.
Checkpoint location stores source offsets so the source is replayable and batches are reconstructed identically.

Question. Use Spark Structured Streaming to upsert a stream into a Delta table exactly-once, given that foreachBatch may re-run a batch on recovery.

Input.

Fact	Value
Source	Kafka topic `events`
Sink	Delta table `events_delta` keyed on `event_id`
Hazard	`foreachBatch` may re-invoke the same `batchId`
Goal	each `event_id` upserted exactly once

Code.

from delta.tables import DeltaTable

def upsert_to_delta(microBatchDF, batchId):
    # De-dup within the batch, then MERGE on the business key (idempotent by construction)
    deduped = microBatchDF.dropDuplicates(["event_id"])
    (DeltaTable.forName(spark, "events_delta").alias("t")
        .merge(deduped.alias("s"), "t.event_id = s.event_id")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        # Delta ignores a re-applied (appId, version) => skip-on-replay for free:
        .execute())

(spark.readStream.format("kafka")
     .option("subscribe", "events").load()
     .selectExpr("CAST(value AS STRING)")   # ... parse to columns incl. event_id ...
     .writeStream
     .foreachBatch(upsert_to_delta)
     .option("checkpointLocation", "/chk/events_delta")  # replayable offsets
     .trigger(processingTime="30 seconds")
     .start())

Step-by-step trace.

Spark reads Kafka, forms micro-batch 128, and calls upsert_to_delta(df, 128); the MERGE on event_id inserts new events and updates existing ones.
The job crashes after the MERGE commits but before Spark records batch 128 as done in its checkpoint.
On restart Spark reconstructs the identical batch 128 from the checkpointed Kafka offsets and re-invokes upsert_to_delta(df, 128).
The re-run MERGE matches the rows it already wrote and overwrites them with identical values — idempotent, so no duplicates (and Delta's transaction log would also skip a re-applied version).
Because the MERGE key is the business event_id, the write is exactly-once regardless of how many times the batch is retried.

Output:

Delivery of batch 128	Rows for a given `event_id`
first `MERGE`	1
replay `MERGE` (same batchId)	1 (overwritten)
any further replay	1 (idempotent)

Rule of thumb. In foreachBatch, always make the sink write idempotent — MERGE on a business key or skip-on-batchId — because Spark can and will re-run the same batchId after a failure.

Interview scenario on writing to a non-transactional store

You must stream aggregated results from Flink into an object store (or a database with no cross-request transaction you control) exactly-once. Direct writes duplicate on recovery because the store is outside Flink's checkpoint. Design the sink.

Solution Using a staged-then-commit two-phase-commit sink with idempotent commit

Answer choices.

A. Write each record directly to the store as it is processed; accept occasional duplicates as "eventually consistent."
B. Buffer per checkpoint, stage on preCommit, and atomically commit (rename / commit-by-id) only after the checkpoint completes, with an idempotent commit for replay safety.
C. Write directly but run a downstream dedup job over the store every hour.
D. Disable checkpointing so there is no replay to cause duplicates.

Code.

Elimination:
A  direct eager writes    -> replay after recovery re-writes -> duplicates          [reject]
C  hourly dedup job        -> duplicates visible for up to an hour; O(store) cleanup  [reject: not exactly-once]
D  disable checkpointing   -> now there is no fault tolerance at all -> data loss     [reject: worse]
B  2PC sink: stage on preCommit, idempotent commit on checkpoint-complete            [ACCEPT]

Step-by-step trace.

Constraints: the store is outside the engine's checkpoint, and recovery replays post-checkpoint records — direct writes therefore duplicate.
A accepts duplicates, which fails "exactly-once" outright — eliminate.
C papers over duplicates with a periodic cleanup — they are still visible for up to an hour and the cleanup is O(store) — eliminate.
D removes checkpointing, which removes fault tolerance entirely: a crash now loses in-flight state — strictly worse — eliminate.
B ties the external commit to the checkpoint: records stage on preCommit (durable, invisible), the commit finalises only after the checkpoint completes, and because the commit is idempotent (atomic rename / commit-by-id), a recovery that re-issues it does not duplicate.

Output:

Requirement	Mechanism
No duplicates on replay	commit only after checkpoint completes
Safe to recover	idempotent commit (rename / commit-by-id)
No data loss	checkpointing retained; abort discards partials
No cleanup pass	correctness is structural, not compensating

Why this works — concept by concept:

Checkpoint-coordinated commit — deferring the external commit until the checkpoint completes means only records whose state and offsets are durably captured become visible, aligning the sink with the engine's fault-tolerance boundary.
Pre-commit / commit split — staging in preCommit makes data durable without exposing it, so a failed checkpoint's data is aborted rather than leaked.
Idempotent commit — the atomic rename / commit-by-id makes re-issuing the commit after a mid-commit crash a no-op, which is the property that closes the last failure window.
Cost — output is delayed by up to one checkpoint interval and each checkpoint carries staging overhead, so you trade a bounded latency and I/O cost for end-to-end exactly-once to a store that has no native transaction.

Streaming
Topic — streaming
Checkpointing and transactional-sink problems

Practice →

Data processing Topic — data-processing Fault-tolerant batch-and-stream sink problems

Practice →

5. End-to-end EOS across source → process → sink; failure & replay

Exactly-once is a property of the whole chain — replayable source, checkpointed state, transactional sink

The invariant: end-to-end exactly once streaming holds only when all three links hold simultaneously — a replayable source that can be re-read from a durable position, processing state that is checkpointed atomically with that position, and a sink that is transactional or idempotent — and the guarantee of the whole pipeline is the weakest of the three, so an exactly-once engine feeding an append-only sink is only at-least-once end to end. You do not "turn on" exactly-once; you verify the chain.

The three links, restated for the whole pipeline.

Replayable source. On recovery the pipeline must resume from a committed position and re-deliver everything after it: Kafka offsets, Kinesis sequence numbers, a file's byte offset, a database WAL/CDC LSN. If the source cannot be re-read, dropped records are gone — you are capped at at-most-once.
Checkpointed state + offsets, atomically. The engine periodically snapshots operator state and the source position into one consistent checkpoint. Atomicity is essential: if state and offset could diverge, recovery would either double-count (offset behind state) or lose (offset ahead of state).
Transactional or idempotent sink. The replayed writes after recovery must not duplicate — via a two-phase-commit sink tied to the checkpoint, or an idempotent upsert keyed on a business id. This is where sections 2 and 4 plug in.

Checkpointing / snapshotting — how the middle link works. Flink's mechanism (Chandy–Lamport-style distributed snapshots) injects checkpoint barriers into the stream. Each operator, on receiving barriers on all its inputs (barrier alignment), snapshots its state; when every operator has snapshotted, the checkpoint is complete and the source offsets it captured are the committed recovery position. Aligned checkpoints give exactly-once (an operator waits for all barriers so its snapshot is consistent); unaligned checkpoints trade some of that for lower latency under backpressure. Spark's equivalent is the checkpoint directory that records source offsets and state per micro-batch. Setting the mode to EXACTLY_ONCE (vs AT_LEAST_ONCE, which skips alignment) is the switch.

The failure-and-replay walkthrough — what actually happens on a crash. This is the story senior candidates can tell end to end:

The job crashes mid-window; some records after the last completed checkpoint were processed and possibly staged at the sink, but the checkpoint they belong to never completed.
Recovery restores operator state and source offsets from the last completed checkpoint — both rewind together to a consistent point.
The pipeline replays every record after that offset; the state is exactly what it was at the checkpoint, so reprocessing produces the same results.
The sink either aborts the never-committed staged transaction and re-stages (2PC), or receives the replayed writes and upserts them onto the same keys (idempotent) — so the replay produces no duplicate at the sink.

Interview signals — what a strong answer contains. When an interviewer asks "how would you make this pipeline exactly-once," the tells that you actually understand it:

You name all three links and say the guarantee is the weakest link — not "I'll enable exactly-once mode."
You distinguish delivery (at-least-once transport) from effect (deduplicated / transactional), and use "effectively-once."
You mention the idempotent-commit requirement on recovery, and the two-generals caveat (you cannot avoid duplicate delivery, only duplicate effect).
You acknowledge the cost — latency added by commit/checkpoint intervals, throughput lost to coordination — and when at-least-once + idempotency is the pragmatic choice instead.

Common traps to pre-empt.

Exactly-once engine + append-only sink — the sink caps you at at-least-once. Add idempotency or a 2PC sink.
Non-replayable source — caps you at at-most-once; no downstream mechanism recovers a dropped input.
Assuming state and offsets are independently committed — they must be one atomic checkpoint, or recovery double-counts or loses.
Ignoring checkpoint interval as a latency floor — output only becomes visible on checkpoint/commit, so the interval sets end-to-end latency.

End-to-end Flink Kafka-to-Kafka exactly-once — a worked teaching example

Detailed explanation. Assemble the whole chain in Flink: a Kafka source (replayable offsets), a keyed stateful aggregation (checkpointed), and an exactly-once Kafka sink (two-phase commit via Kafka transactions). The three config decisions that make it exactly-once are enableCheckpointing with EXACTLY_ONCE mode (atomic state+offset snapshots), a Kafka source whose offsets are committed as part of the checkpoint (not auto-committed), and a Kafka sink at DeliveryGuarantee.EXACTLY_ONCE whose transaction commits on checkpoint completion. Get all three and a crash replays cleanly with no duplicate on the output topic.

Source: Kafka connector commits offsets in the checkpoint, so state and position rewind together.
State: keyed aggregate snapshotted on each checkpoint barrier (aligned = exactly-once).
Sink: EXACTLY_ONCE Kafka sink pre-commits on snapshot, commits on checkpoint-complete.
transaction.timeout.ms > checkpoint interval so a staged transaction survives until its checkpoint commits.

Question. Configure a Flink Kafka→Kafka job (windowed count) to be exactly-once end to end.

Input.

Link	Choice
Source	Kafka source, offsets committed on checkpoint
State	keyed window count, checkpointed
Sink	Kafka sink, `DeliveryGuarantee.EXACTLY_ONCE`
Safety	`transaction.timeout.ms` > checkpoint interval

Code.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60_000);                                   // snapshot state + offsets / 60s
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

KafkaSource<Event> source = KafkaSource.<Event>builder()
    .setBootstrapServers("broker:9092")
    .setTopics("clicks").setGroupId("agg")
    .setStartingOffsets(OffsetsInitializer.committedOffsets())     // replayable position
    .setDeserializer(new EventDeserializer())
    .build();   // offsets are committed as part of Flink's checkpoint, not auto-commit

KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("broker:9092")
    .setRecordSerializer(/* topic "click-counts" */ serializer)
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)          // Kafka txn per checkpoint
    .setTransactionalIdPrefix("flink-clickcounts")
    .setProperty("transaction.timeout.ms", "900000")              // > checkpoint interval
    .build();

env.fromSource(source, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)), "kafka")
   .keyBy(e -> e.userId)
   .window(TumblingEventTimeWindows.of(Time.minutes(1)))
   .aggregate(new CountAgg())                                      // checkpointed keyed state
   .map(Object::toString)
   .sinkTo(sink);

Step-by-step trace.

The Kafka source reads clicks; its offsets are captured inside each Flink checkpoint, so the source position is always consistent with the aggregation state.
The keyed 1-minute window count builds state; on each checkpoint barrier the operator snapshots that state atomically with the captured offsets.
The EXACTLY_ONCE Kafka sink opens a Kafka transaction and pre-commits it on the snapshot; the produced click-counts records are staged but invisible to read_committed consumers.
When the checkpoint completes globally, Flink commits the sink's Kafka transaction, making that window's output visible exactly once.
On a crash, Flink restores state and offsets from the last completed checkpoint and replays; the never-committed sink transaction is aborted, so the output topic shows each window's count once.

Output:

Event	State + offset	Output topic
checkpoint N completes	committed together	window counts visible once
crash after N, before N+1	restore to N, replay	aborted staged txn, no dup
downstream `read_committed`	—	sees only committed counts

Rule of thumb. End-to-end Flink EOS = checkpointing in EXACTLY_ONCE mode + a Kafka source whose offsets ride the checkpoint + an EXACTLY_ONCE Kafka sink with transaction.timeout.ms larger than the checkpoint interval — miss any one and you drop to at-least-once.

Failure-and-replay trace with no duplicate at the sink — a worked teaching example

Detailed explanation. The single most convincing thing you can demonstrate is a concrete crash-and-replay walkthrough that shows the sink emitting each result exactly once despite the record being delivered twice. Take a pipeline counting events per key, checkpoint every N records, and crash between two records after a checkpoint. The point is to show that state and offset rewind together to the last completed checkpoint and that the sink's idempotent/transactional commit absorbs the replayed writes.

Before crash: checkpoint at offset 100 captured count{A}=5; records 101–103 processed, sink staged but not committed.
Crash at record 103.
Restore: state count{A}=5 and offset 100 come back together (consistent).
Replay: records 101–103 reprocessed → count{A} recomputed identically; sink aborts the old staged transaction and commits the new one → one result.

Question. Show that a crash after a checkpoint replays without a duplicate at the sink for an event-count pipeline.

Input.

Item	Value
Aggregation	count events per `key`
Last checkpoint	offset 100, state `count{A}=5`
Records after checkpoint	101,102,103 (all key A), then crash
Sink	transactional (commit on checkpoint)

Code.

Timeline (offset : action)
100 : CHECKPOINT completed  -> state{A}=5 committed WITH offset=100
101 : process A -> state{A}=6 ; sink stage (uncommitted txn T1)
102 : process A -> state{A}=7 ; sink stage (T1)
103 : process A -> state{A}=8 ; *** CRASH *** (T1 never committed)

--- recovery ---
restore    : state{A}=5 , offset=100     # rewound TOGETHER to last completed checkpoint
abort T1   : staged 101-103 output discarded (never was visible)
replay 101 : state{A}=6 ; stage into fresh txn T2
replay 102 : state{A}=7 ; T2
replay 103 : state{A}=8 ; T2
CHECKPOINT : commit T2 -> sink shows count{A}=8 exactly once

Step-by-step trace.

At offset 100 the checkpoint commits state count{A}=5 together with the offset, establishing a consistent recovery point.
Records 101–103 advance the in-memory state to 8 and stage output into transaction T1, which is not yet committed.
The crash at 103 leaves T1 uncommitted; because state and offset were only durably committed at offset 100, recovery rewinds both to that point.
Recovery aborts T1 (its staged output was never visible), then replays 101–103, recomputing count{A}=8 deterministically from the restored state.
The next checkpoint commits transaction T2, so the sink publishes count{A}=8 exactly once — the records were delivered twice (101–103 processed before and after the crash) but the effect is single.

Output:

At the sink	Value
before crash (T1)	nothing visible (uncommitted)
after replay (T2 committed)	`count{A}=8` once
duplicate rows	none

Rule of thumb. Exactly-once is provable by this walkthrough: state and offset must rewind together, the pre-crash sink transaction must abort, and the replayed output must commit idempotently — if you can narrate those three, you understand end-to-end EOS.

Interview scenario on designing an end-to-end exactly-once pipeline

Design a payments pipeline: ingest transfer events, compute per-account running balances with windowed fraud checks, and write authoritative results to a downstream store — end-to-end exactly-once, because a duplicated or dropped transfer is a financial incident. Name the mechanism at each link.

Solution Using a replayable source + checkpointed state + transactional/idempotent sink

Answer choices.

A. Kafka source with EOS transactions + Flink checkpointed state + a two-phase-commit (or idempotent-upsert) sink, all read_committed downstream.
B. Any source + exactly-once engine + an append-only results table (rely on the engine's "exactly-once mode").
C. A non-replayable socket source + checkpointed state + transactional sink.
D. Kafka source + stateless direct writes + a nightly dedup/reconciliation job.

Code.

Elimination:
B  append-only sink       -> sink is the weakest link -> at-least-once end to end     [reject]
C  non-replayable source  -> dropped inputs unrecoverable -> at-most-once cap          [reject]
D  direct writes + nightly dedup -> duplicates visible for hours; compensating not EOS [reject]
A  replayable source + checkpointed state + transactional/idempotent sink              [ACCEPT]

Step-by-step trace.

Constraints: financial data, so neither loss nor duplication is acceptable — the guarantee must hold at every link.
B pairs an exactly-once engine with an append-only sink; the sink cannot dedup replays, so end-to-end it is only at-least-once — eliminate on the weakest-link rule.
C's socket source is not replayable; a dropped transfer can never be re-read, capping the pipeline at at-most-once regardless of the sink — eliminate.
D writes directly and cleans up nightly — duplicates are visible and acted on for hours, and reconciliation is compensating, not exactly-once — eliminate.
A holds all three links: Kafka offsets make the source replayable, Flink checkpoints state atomically with those offsets, and a 2PC/idempotent sink absorbs the replay — with read_committed downstream so nothing aborted is ever seen.

Output:

Link	Mechanism	Guarantee contributed
Source	Kafka offsets, committed in checkpoint	replayable
Process	Flink `EXACTLY_ONCE` checkpointed state	consistent recovery
Sink	2PC sink / idempotent upsert on `transfer_id`	no duplicate write
Downstream	`read_committed`	never sees aborted output

Why this works — concept by concept:

Weakest-link reasoning — the design is chosen so every link is exactly-once, because the pipeline guarantee is the minimum over links; a single append-only or non-replayable component would silently downgrade the whole thing.
Atomic state-and-offset checkpoint — snapshotting the running balances together with the source offset means recovery cannot double-count or drop a transfer, which is the core correctness property for balances.
Transactional/idempotent sink — tying the external commit to the checkpoint (or upserting on transfer_id) makes the replayed writes harmless, closing the loop from delivery to durable effect.
Cost — coordination adds latency of about one checkpoint interval and some throughput overhead; for money movement that is the correct trade, whereas high-volume telemetry might accept at-least-once + idempotency to shave the cost.

Streaming
Topic — streaming
End-to-end exactly-once pipeline problems

Practice →

Course Course — Spark internals Apache Spark internals for data engineering interviews

Practice →

Cheat sheet — exactly-once recipes

Delivery semantics → what happens to a record (memorise this table).

Guarantee	Duplicates	Data loss	How you get it	When to use
At-most-once	no	yes	commit position before doing the work	lossy telemetry, superseded gauges
At-least-once	yes	no	do the work, then commit position	default; safe with an idempotent sink
Exactly-once (effectively-once)	no	no	at-least-once + idempotent or transactional effect	money, ledgers, audited data

Kafka EOS config block.

# Producer
enable.idempotence = true            # PID + per-partition sequence numbers (default 3.0+)
acks = all                           # commit after all ISR replicas
transactional.id = <stable-per-instance>   # transactions + zombie fencing
# Consumer (source of a read-process-write)
enable.auto.commit = false           # producer commits offsets, transactionally
isolation.level = read_committed     # never read aborted/uncommitted records
# Loop: initTransactions -> beginTransaction -> send... -> sendOffsetsToTransaction -> commitTransaction

Flink / Spark checkpoint + sink config block.

# Flink
env.enableCheckpointing(interval)
CheckpointingMode.EXACTLY_ONCE                 # aligned barriers, atomic state+offset
KafkaSink DeliveryGuarantee.EXACTLY_ONCE       # 2PC via Kafka txn, commit on checkpoint
transaction.timeout.ms > checkpoint interval   # staged txn must survive to commit
# Spark Structured Streaming
option("checkpointLocation", path)             # replayable offsets + state
foreachBatch((df, batchId) -> idempotent upsert)   # MERGE on business key / skip-on-batchId
# built-in file sink + _spark_metadata is exactly-once by batchId

Idempotency vs transaction — the decision line. Sink supports an upsert on a stable business key → idempotency (cheapest; no coordinator). Sink write must be atomic with the offset/checkpoint or spans multiple systems/partitions → transaction / two-phase commit. Reach for 2PC only when idempotency cannot express the write.

Failure mode → guarantee it needs.

Failure	Fixed by
Producer retry after lost ack	idempotent producer (PID + seq)
Consumer reprocesses after rebalance	transaction with `sendOffsetsToTransaction` + fencing
Engine replays after crash	checkpointed state+offset + idempotent/2PC sink
External sink re-written on recovery	pre-commit/commit tied to checkpoint, idempotent commit
Downstream sees aborted output	`isolation.level = read_committed`

Interview one-liners. "Exactly-once is effectively-once: at-least-once delivery + deduplicated/transactional effect." · "The pipeline guarantee is the weakest of source, state, and sink." · "Two-phase commit stages on the checkpoint and commits when it completes — and the commit must be idempotent." · "Kafka transactions cover Kafka; an external sink needs its own 2PC or idempotency."

Frequently asked questions

Is exactly-once really possible, or is it a myth?

It is real, but the name is misleading — the honest term is effectively-once. The network cannot deliver a message precisely once (the two-generals problem forces retries on ambiguous acknowledgements), so a record will sometimes be delivered more than once. What real systems guarantee is that the duplicate delivery has no duplicate effect, achieved through idempotency or transactions. So exactly-once semantics is a statement about the observable outcome, not about the wire.

What is the difference between at-least-once and exactly-once?

at-least-once vs exactly-once comes down to what happens to duplicates. At-least-once never loses a record but may deliver it more than once, so your sink can see the same event twice. Exactly-once builds on the same at-least-once delivery but makes those duplicates harmless — either the write is idempotent (a replay overwrites the same row) or the write commits transactionally with the source offset (a replay is rolled back) — so the net effect is one. In short: at-least-once is a transport property; exactly-once is an effect property layered on top of it.

How does Kafka achieve exactly-once semantics?

kafka exactly once combines three things. The idempotent producer stamps records with a producer id and per-partition sequence numbers so the broker drops retried duplicates. Transactions (a stable transactional.id, beginTransaction/commitTransaction, and sendOffsetsToTransaction) make the produced output and the consumed offsets commit atomically, with zombie fencing on rebalance. Finally, downstream consumers set isolation.level=read_committed so they never read aborted or uncommitted records — together these give the atomic read-process-write that defines Kafka EOS.

Idempotency or transactions — which should I use for a sink?

Use idempotency whenever the sink supports an upsert on a stable business key — it is the cheapest exactly-once mechanism because it needs no coordinator, no distributed transaction, and no two-phase protocol; a replay simply overwrites the same row. Reach for a transactional sink or a two-phase commit only when idempotency cannot express the write: when the sink write must be atomic with the source offset or checkpoint, or when it spans multiple partitions or systems that must commit all-or-nothing. Many production pipelines are deliberately at-least-once plus an idempotent sink for exactly this reason.

What does two-phase commit have to do with streaming sinks?

An external sink (a database, an object store) is not covered by the streaming engine's checkpoint, so its write must be coordinated with the checkpoint through a two-phase commit. Phase one (pre-commit) stages the write durably but invisibly when the operator snapshots for a checkpoint; phase two (commit) makes it visible only after the whole checkpoint completes. Flink's TwoPhaseCommitSinkFunction implements exactly this, and the commit step must be idempotent so a crash between checkpoint-complete and commit can safely re-run it.

Does exactly-once slow the pipeline down?

Yes, modestly, and the trade is usually worth it for critical data. Transactions add coordinator round-trips and hold output invisible until commit, and checkpointing adds periodic snapshot overhead plus a latency floor — output only becomes visible at each checkpoint/commit interval, so end-to-end latency rises by roughly that interval. For money movement and audited data that cost is correct; for high-volume telemetry where a small duplicate rate is harmless, at-least-once plus a lightweight idempotent sink is often the better engineering trade.

Practice on PipeCode

Turn exactly-once theory into a reflex

Blog posts explain the guarantees. PipeCode drills build the instinct interviews and design reviews actually test — naming the weakest link, choosing idempotency versus a transaction, and narrating a crash-and-replay that leaves no duplicate at the sink. Pipecode.ai is Leetcode for Data Engineering — scenario-first practice on streaming, ETL, and event processing tuned to the delivery-semantics trade-offs real pipelines demand.

Practice streaming problems →
Practice ETL problems →

DEV Community

Exactly-Once Semantics in Streaming: Idempotency, Transactions & Two-Phase-Commit Sinks

1. Delivery semantics — at-most-once, at-least-once, exactly-once

There are only three guarantees, and they differ in whether a record can be dropped, duplicated, or neither

Worked example — tracing one record through a crash under each semantic

2. Idempotency & dedup — the cheapest path to effectively-once

Make the write harmless to repeat, and at-least-once delivery becomes exactly-once for free

Idempotent upsert sink keyed on event_id — a worked teaching example

Kafka idempotent producer — sequence numbers reject the retry — a worked teaching example

Interview scenario on deduplicating an at-least-once stream

Solution Using a MERGE upsert on the event_id dedup key

3. Transactions — Kafka EOS, transactional producer/consumer, read_committed

When the write must be atomic with the offset, wrap read-process-write in one transaction

The exactly-once consume-transform-produce loop — a worked teaching example

Turning on framework-level EOS — Kafka Streams and Flink config — a worked teaching example

Interview scenario on a money-movement stream

Solution Using a transactional read-process-write with sendOffsetsToTransaction

4. Two-phase-commit / transactional sinks — checkpoint-coordinated commit

When the sink lives outside the engine, commit it in two phases tied to the checkpoint

A Flink two-phase-commit file sink — stage then atomically rename — a worked teaching example

Spark Structured Streaming idempotent sink on batchId — a worked teaching example

Interview scenario on writing to a non-transactional store

Solution Using a staged-then-commit two-phase-commit sink with idempotent commit

5. End-to-end EOS across source → process → sink; failure & replay

Exactly-once is a property of the whole chain — replayable source, checkpointed state, transactional sink

End-to-end Flink Kafka-to-Kafka exactly-once — a worked teaching example

Failure-and-replay trace with no duplicate at the sink — a worked teaching example

Interview scenario on designing an end-to-end exactly-once pipeline

Solution Using a replayable source + checkpointed state + transactional/idempotent sink

Cheat sheet — exactly-once recipes

Frequently asked questions

Is exactly-once really possible, or is it a myth?

What is the difference between at-least-once and exactly-once?

How does Kafka achieve exactly-once semantics?

Idempotency or transactions — which should I use for a sink?

What does two-phase commit have to do with streaming sinks?

Does exactly-once slow the pipeline down?

Practice on PipeCode

Turn exactly-once theory into a reflex

Top comments (0)