Exactly-once semantics

Active contributors: David Jacot, Apoorv Mittal, Lianet Magrans, Justine Olshan, Matthias J. Sax

Kafka offers exactly-once semantics (EOS) on the producer side via two related mechanisms:

1. Idempotent produce — at-most-once delivery of each batch even under retries; on by default.
2. Transactional produce — a producer can atomically write to multiple partitions and commit consumer offsets in the same transaction (read-process-write pipelines). Kafka Streams (processing.guarantee=exactly_once_v2) and Connect source connectors (KIP-618) build on this.

The protocol was introduced in KIP-98 (Kafka 0.11) and has been refined many times since, most recently with the producer ID rotation (KIP-360) and improvements to abort handling. This page is the runtime view; the source pointers are the easiest path into the code.

Components

graph TD
    P[KafkaProducer] -->|InitProducerId| TC[Transaction coordinator]
    P -->|AddPartitionsToTxn| TC
    P -->|AddOffsetsToTxn| TC
    P -->|Produce| L[Leader broker]
    L -->|append, validate seq| Log[(UnifiedLog + ProducerStateManager)]
    P -->|EndTxn commit/abort| TC
    TC -->|WriteTxnMarkers| Lleaders[All leaders involved]
    Lleaders -->|append commit/abort marker| Log
    Cons[Consumer read_committed] --> L
    L --> Cons

Concern	Where
Producer-side state machine	clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
Producer-side public API	clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java (initTransactions, beginTransaction, sendOffsetsToTransaction, commitTransaction, abortTransaction)
Broker-side append validation	storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java
Transaction coordinator (legacy)	core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala
Transaction coordinator (Java port)	transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/
Aborted-transaction index	storage/.../log/TransactionIndex.java
Streams EOS	streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java (transactional Producer per task)
Connect source EOS (KIP-618)	connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java

Idempotent producer

When enable.idempotence=true (default since 3.0), every KafkaProducer:

1. Calls InitProducerId once at startup; the transaction coordinator returns a (producerId, epoch).
2. Tags each RecordBatch with (producerId, epoch, baseSequence).
3. Per partition, sends batches with monotonically increasing baseSequence.

The leader's ProducerStateManager keeps the latest sequence per (partition, producerId). On a duplicate (re-sent batch, same sequence) it returns success without re-appending. On an out-of-order sequence it returns OutOfOrderSequenceException. This guarantees per-partition de-duplication across retries.

KIP-360 added per-producer-ID rotation: when a producer's epoch is fenced (e.g., by a transaction abort or a long pause), the producer can InitProducerId again with the same transactional.id and resume; older code would error out.

Transactional produce

Transactions extend the idempotent model across multiple partitions and commit topics atomically.

sequenceDiagram
    participant App
    participant P as KafkaProducer (TransactionManager)
    participant TC as Transaction coordinator
    participant L1 as Leader (topic A)
    participant L2 as Leader (topic B)
    App->>P: initTransactions()
    P->>TC: InitProducerId(transactional.id)
    TC-->>P: (pid, epoch)
    App->>P: beginTransaction()
    App->>P: send(record-on-A); send(record-on-B)
    P->>TC: AddPartitionsToTxn(A, B)
    P->>L1: ProduceRequest (with isTransactional=true)
    P->>L2: ProduceRequest
    App->>P: sendOffsetsToTransaction(consumerOffsets, groupMetadata)
    P->>TC: AddOffsetsToTxn(__consumer_offsets)
    P->>L_offsets: TxnOffsetCommitRequest
    App->>P: commitTransaction()
    P->>TC: EndTxn(commit)
    TC->>TC: write PrepareCommit marker on __transaction_state
    TC->>L1: WriteTxnMarkers(commit)
    TC->>L2: WriteTxnMarkers(commit)
    TC->>L_offsets: WriteTxnMarkers(commit)
    TC->>TC: write CompleteCommit marker
    TC-->>P: success

A commit writes a control record (commit marker) on every involved partition. A consumer reading with isolation.level=read_committed ignores records past an open transaction's last stable offset until it sees a commit (visible) or abort (skip) marker. The aborted transaction index (.txnindex) lets the broker skip over aborted batches efficiently during fetch.

abort writes abort markers; consumers in read_committed mode skip those records. Since the producer's epoch is bumped on abort, any in-flight batches from before the abort that arrive after the marker are fenced (InvalidProducerEpochException).

Streams exactly_once_v2

Each StreamTask (one per input partition group) gets its own transactional producer with transactional.id = <application.id>-<task.id>. Per commit interval the task:

1. Begins a transaction.
2. Processes records from input partitions, writes to output partitions and changelog topics.
3. Calls producer.sendOffsetsToTransaction(consumerOffsets, groupMetadata) to atomically advance input offsets.
4. Commits.

Crash recovery: when the task moves to a different instance, the producer's old transactional.id is fenced (InitProducerId bumps the epoch); any leftover in-flight records from the previous incarnation are aborted.

exactly_once_v2 (since 3.0) replaced the older exactly_once_beta and exactly_once modes; older modes are removed. The "v2" name reflects that all tasks within an instance share producer thread pools more efficiently.

Connect source EOS (KIP-618)

ExactlyOnceWorkerSourceTask wraps the connector's record producer in a transaction per polled batch:

1. connectorTransactionBoundaries (defined by the connector or the framework) decide where transactions begin/end.
2. The task writes records and source-offset records (which were previously stored externally) inside the same transaction.
3. On commit, the offsets are durable atomically with the records — a crash never leaves duplicate records downstream.

Sink-side EOS depends on the destination system; Connect provides hooks (SinkTask.preCommit) for connectors to integrate with their sinks' transactional semantics.

Failure modes

• OutOfOrderSequenceException — usually means a transaction was aborted (epoch bump fenced subsequent batches) or the producer state was lost on the broker (segment deletion + producer ID expiration). Recovery: close + recreate the producer; resume from the last committed offset.
• ProducerFencedException — a producer with the same transactional.id started elsewhere and bumped the epoch. The current producer must shut down.
• InvalidProducerEpochException — newer variant that allows recovery without recreating the producer (KIP-360).
• UnknownProducerIdException — the broker has no record of this producer ID. Happens when the producer state has been aged out (producer.id.expiration.ms); the broker treats this as a fatal error for the producer.
• Long-running transactions blocking compaction — a transaction that's been open for longer than transaction.max.timeout.ms is auto-aborted by the coordinator. Long-running idempotent-only producers (no transactions) don't have this issue but can be aged out by producer.id.expiration.ms.

Configuration

Producer-side:

• enable.idempotence (default true) — turns on idempotent produce.
• transactional.id — set this to use transactions. Must be stable across restarts of the same logical producer.
• transaction.timeout.ms — server-side cap on open transactions.
• delivery.timeout.ms, retries, max.in.flight.requests.per.connection — interact with retries and ordering.

Streams:

• processing.guarantee=exactly_once_v2.

Connect:

• exactly.once.source.support — disabled, preparing, enabled.

Broker / cluster:

• transactional.id.expiration.ms — how long the coordinator keeps a transactional ID's state after last activity.
• transaction.state.log.replication.factor, transaction.state.log.min.isr — durability of __transaction_state.

Entry points for modification

• Producer state machine: TransactionManager.java plus Sender.java. Almost any change here is observable to clients; KIP required.
• Broker append validation: ProducerStateManager.java. Format-level changes also affect snapshots and recovery.
• Transaction coordinator: state machine in TransactionStateManager.scala (legacy) + transaction-coordinator/. The coordinator is being progressively ported to Java.
• Streams EOS: StreamTask.java and TaskManager.java; commit / suspend / close paths must stay correct under crash.
• Read-committed semantics: the consumer-side filter is in clients/.../consumer/internals/Fetcher.java.

Related pages

• Modules: clients — producer/consumer entry points.
• Modules: coordinators — transaction coordinator.
• Modules: storage — ProducerStateManager, TransactionIndex.
• Modules: streams — task-level EOS.
• Modules: connect — source-task EOS.