Kafka exactly-once, end to end

What it is

Kafka's exactly-once semantics (EOS) is the combination of two independent mechanisms: an idempotent producer, which prevents a retry from writing the same record twice to the same partition, and transactions, which make a set of writes across partitions plus a consumer offset commit atomic. Together they give you exactly-once for the consume, transform, produce loop, entirely inside Kafka.

The phrase is commonly confused with two other things. It is not end-to-end exactly-once across your whole architecture: the moment a record leaves Kafka for a REST call, an email, a payment or a database that is not participating in the transaction, the guarantee stops. And it is not "the message is delivered once". Kafka still delivers at least once at the network level; what changes is that duplicates are recognised and discarded, and that partial writes are never visible to a consumer that asks not to see them. The honest one-liner is that exactly-once is at-least-once delivery plus deduplication plus atomic visibility, and every part of that sentence is doing work.

The problem it solves

Without idempotence, a producer that sends a record, has the broker write it, and then loses the acknowledgement to a network blip will retry. The broker has no way to know the second copy is the same record, so the partition now contains the record twice. Turning retries off is not an option, because then a recoverable blip becomes data loss.

Without transactions, a stream processor that reads from topic A, writes a derived record to topic B and then commits its offset on A has three orderings available and all of them are wrong in some failure. Commit the offset first and a crash loses the output. Write the output first and a crash reprocesses the input, producing a duplicate downstream. Write to two output topics and crash in between, and the two topics permanently disagree. Transactions collapse the write to B, the write to C and the offset commit on A into one atomic unit.

Mechanics

The idempotent producer

Set enable.idempotence=true (the default since Kafka 3.0). On initProducerId, the broker assigns the producer a PID and the producer starts a per-partition sequence number at zero. Every produce request carries (PID, epoch, partition, base_sequence).

The broker keeps the last five sequence numbers per PID per partition in the partition's producer state. On arrival it compares:

Incoming sequence	Broker's state	Result
expected (`last + 1`)	accept, advance	written once
already seen	`DUPLICATE_SEQUENCE_NUMBER`	silently treated as success, no second write
ahead of expected	`OUT_OF_ORDER_SEQUENCE_NUMBER`	rejected; the producer would have created a gap

That five-record window is exactly why max.in.flight.requests.per.connection must be at most 5 with idempotence enabled. It is also why idempotence, contrary to the folklore, does not cost you pipelining: you keep five requests in flight and still get ordering, because the broker rejects anything that would land out of order.

Idempotence is scoped to one producer session and one partition. Restart the process without a transactional.id and you get a fresh PID, so a record in flight during the crash can be written again by the new session. That gap is what transactional.id closes.

Transactions

// Producer config for a consume-transform-produce processor.
//   transactional.id must be STABLE across restarts of this logical processor
//   and UNIQUE per processor instance. It is the identity the broker fences on.
props.put("transactional.id", "orders-enricher-1");
props.put("enable.idempotence", "true");          // implied, but be explicit
props.put("acks", "all");                          // implied by idempotence

// Consumer config: never let the framework commit for us.
props.put("enable.auto.commit", "false");
props.put("isolation.level", "read_committed");    // do not read aborted data

producer.initTransactions();   // registers with the transaction coordinator,
                               // bumps the producer EPOCH, and fences any older
                               // instance still holding this transactional.id

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

    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, String> r : batch) {
            producer.send(new ProducerRecord<>("orders-enriched", r.key(), enrich(r.value())));
            producer.send(new ProducerRecord<>("orders-audit",    r.key(), audit(r.value())));
        }

        // The offset commit rides INSIDE the transaction. This is the whole
        // point: output and input position commit or abort together.
        // Passing groupMetadata (not just groupId) is the KIP-447 form, which
        // lets one producer serve all partitions assigned to this consumer.
        producer.sendOffsetsToTransaction(offsetsOf(batch), consumer.groupMetadata());

        producer.commitTransaction();
    } catch (ProducerFencedException | OutOfOrderSequenceException e) {
        // Another instance took our transactional.id, or state is unrecoverable.
        // There is no recovery: close and let the supervisor restart us.
        producer.close();
        throw e;
    } catch (KafkaException e) {
        producer.abortTransaction();   // consumer will re-poll from the last
                                       // committed offset; no duplicates escape
    }
}

Underneath, the transaction coordinator (a broker, chosen by hashing the transactional.id into a partition of the internal __transaction_state topic) runs a two-phase commit. Phase one appends ONGOING plus the set of partitions touched to __transaction_state. On commit it appends PREPARE_COMMIT, then writes a control record (a commit marker) into every data partition the transaction touched, then appends COMPLETE_COMMIT. Those control markers are what a consumer uses to decide what is visible.

A read_committed consumer never reads past the last stable offset (LSO), which is the offset of the earliest still-open transaction. It buffers records belonging to open transactions and, on seeing an abort marker, drops them using the aborted-transaction index the broker returns with the fetch.

That LSO rule has a consequence worth saying out loud: an open transaction blocks read_committed consumers on that partition, for every record after it. A stuck processor with a 15 minute transaction.timeout.ms stalls its consumers for 15 minutes. Keep transaction timeouts short (the default is 60 seconds; the broker caps it with transaction.max.timeout.ms, default 15 minutes) and keep transactions short.

A worked example

Take a processor reading orders (6 partitions, 5,000 records/second) and writing to orders-enriched and orders-audit.

The processor polls 500 records, sends 1,000 records across the two output topics, sends offsets, and commits. Suppose the machine loses power immediately after the last send and before commitTransaction.

The coordinator's timer expires (transaction.timeout.ms, say 30s) and it aborts: it writes abort markers into every partition of both output topics that the transaction touched.
Consumers of orders-enriched running read_committed have been holding those 500 records in a buffer since they arrived, and have not advanced their LSO past them. On the abort marker they discard the buffer. Nothing downstream ever saw them.
A supervisor restarts the processor with the same transactional.id. initTransactions() bumps the epoch, which fences the dead instance permanently: if it comes back from a long GC pause and tries to commit, it gets ProducerFencedException.
The consumer's committed offset on orders was never advanced, so the new instance re-polls the same 500 records and reprocesses them. The outputs are written a second time, under a new transaction, and this time committed. Downstream sees each record exactly once.

The cost, measured: each commit is two appends to __transaction_state plus one control record per touched partition. Confluent's original benchmarking of the feature reported roughly a 3 percent throughput reduction when transactions commit at 100 ms intervals with 1 KB records, and a much larger penalty for very short transactions, because the fixed cost per commit is amortised over fewer records. The practical tuning knob is therefore batch size, not the feature flag: commit every 100 ms or every few hundred records, not every record.

Production evidence

Kafka Streams is the largest deployment of this machinery. Setting processing.guarantee=exactly_once_v2 makes every Streams task run the loop above automatically, including for its internal state store changelogs, which is why a Streams application can restore a RocksDB state store after a crash and have it agree exactly with the output topic. exactly_once_v2 (Kafka 2.6+, the only supported form since 3.0 deprecated the original) is the KIP-447 implementation, which cut the number of producers needed from one per input partition to one per instance.

Apache Flink's Kafka sink implements exactly-once by mapping Flink's checkpoint barriers onto Kafka transactions: it begins a transaction after each checkpoint, and commits it when the checkpoint completes. This is the canonical example of Kafka transactions being used as the commit half of an external two-phase commit, and it is the mechanism behind Flink's end-to-end exactly-once claim for Kafka-to-Kafka pipelines.

Debezium deliberately does not rely on it for source connectors. Change data capture emits at least once and expects idempotent consumers, because the source of truth is a database transaction log that Kafka cannot enlist in. That choice is worth citing, because it shows a mature system declining the feature where it does not fit.

The debate

The alternative, and in my experience the more common production choice, is at-least-once delivery plus idempotent consumers. You keep enable.idempotence=true for producer-side dedupe, commit offsets after processing, and make every side effect idempotent: upsert by key instead of insert, carry a natural or synthetic idempotency key, and let the sink dedupe.

Choose Kafka transactions when the entire pipeline is Kafka to Kafka, the processing is stateful enough that reconstructing a dedupe key is awkward, and you are already running Kafka Streams or Flink, which implement the hard parts for you. Choose at-least-once plus idempotent sinks when any hop leaves Kafka, when your sinks are databases or HTTP services (they are), or when you cannot tolerate a stuck transaction blocking read_committed consumers.

Exactly-once is the wrong choice when the processor's output goes to an external system: the transaction cannot span it, so you need an idempotency key at that boundary anyway. Once you have built that, the Kafka-internal transaction is buying you very little. It is also wrong when your consumers cannot run read_committed, because a read_uncommitted consumer sees aborted records and the whole scheme is decoration.

The staff-level position: use enable.idempotence everywhere, use transactions only where the pipeline is closed under Kafka, and design every external side effect to be idempotent regardless. The idempotency key at the boundary is not a fallback for when transactions fail, it is the actual guarantee; transactions are an optimisation that removes duplicate work inside the pipeline.

Follow-up Q&A

"What exactly does exactly-once not cover?" Anything outside Kafka. A consumer that reads a committed record and calls a payment API can crash after the call and before its offset commit; on restart it calls the API again. It also does not cover consumer-side processing that is not part of a transaction, does not survive a read_uncommitted consumer, and does not deduplicate records that your producer genuinely sent twice at the application level (two clicks, two HTTP retries from a client). Application-level duplicates need an application-level idempotency key.

"Why does max.in.flight.requests.per.connection have to be 5 or less?" Because the broker only remembers the last five sequence numbers per producer per partition. With more requests in flight, a retry of an old request can arrive after the window has moved past it, and the broker can no longer tell whether it is a duplicate or a gap. It errs on the side of safety and rejects.

"What happens if the same transactional.id is used by two instances?" The second one to call initTransactions() wins. It receives a higher epoch, and the coordinator refuses any further request from the older epoch with ProducerFencedException. This is the zombie-fencing property, and it is why the transactional.id must be stable per logical processor and unique per instance: derive it from something like ${app}-${task-id}, never from a random UUID (which makes fencing impossible) and never from a shared constant (which makes two healthy instances fence each other in a loop).

"A consumer is stuck and lag is climbing, but the processor looks healthy. What do you check?" The last stable offset. If an upstream transactional producer has an open transaction, read_committed consumers cannot advance past it, so lag climbs while error rates stay flat. Check kafka.server:type=BrokerTopicMetrics alongside the producer's transaction state, and look for a producer stuck between beginTransaction and commitTransaction. The fix is a shorter transaction.timeout.ms, which trades a longer stall for a bounded one.

"How do you migrate an existing at-least-once pipeline to exactly-once?" You do not flip it on globally. Enable idempotence first, which is safe and nearly free. Then convert one processor at a time, and only processors whose output topics have consumers you control and can switch to read_committed. Watch consumer lag and p99 on those consumers for a full traffic cycle before converting the next one, because the LSO stall is a new failure mode you have not operated before.

Common misconceptions

The most costly one is believing that enable.idempotence=true alone gives exactly-once. It gives per-partition, per-session deduplication of producer retries, which is genuinely valuable and should be on everywhere, but a producer restart creates a new PID and the guarantee resets. Only a stable transactional.id survives a restart.

The second is believing read_committed is a performance setting. It changes correctness and it changes latency: records are not visible until their transaction commits, so a consumer's end-to-end latency includes the producer's commit interval. If you set a 500 ms commit interval, you have added up to 500 ms of latency to every downstream consumer, and no amount of consumer tuning recovers it.

Interview delivery note

Say this, close to verbatim: "Exactly-once in Kafka is at-least-once plus deduplication plus atomic visibility, and it is scoped to Kafka. The idempotent producer dedupes retries by PID and sequence number; transactions make the output writes and the offset commit atomic; read_committed consumers respect the last stable offset. The moment I write to something that is not Kafka, I need an idempotency key at that boundary, so I design for that first and treat transactions as an optimisation."

The depth signal that separates senior from staff here is naming the last stable offset stall unprompted. Almost every candidate can describe the producer and the transaction; very few have operated a pipeline where an open transaction froze a downstream consumer group, and mentioning it is the fastest way to signal that you have.

The Staff & Team Lead Interview Reference