Kafka Exactly Once Semantics: Idempotent Producers and Transactions

What exactly-once means in Kafka and why duplicates happen

In Kafka, "exactly once" is a narrow guarantee. For a read-process-write loop that consumes from Kafka topics and produces to Kafka topics, each input record's effect appears in the output exactly once, and the consumer's offset advances in step with that output. It does not cover a database write, an HTTP call, or an email sent from inside the loop. Those side effects sit outside the protocol.

Duplicates come from two independent failure points, and each one needs its own fix.

Producer retry after a lost ack. The broker appends the record, but the acknowledgement never reaches the producer because of a timeout or a dropped connection. The producer retries, and the partition now holds two copies. The data did not fail to arrive. The producer could not tell that it had.

Consumer crash before the offset commit. The application processes a record and writes its output, then dies before committing the input offset. On restart it resumes from the last committed offset, reprocesses the record, and emits the output again.

The idempotent producer only prevents duplicate appends within a partition, so it cannot fix the second case. Transactions solve it by committing the output records and the consumer offsets as one atomic unit.

These mechanisms matter to teams that run stream processors computing aggregates, balances, or counts, where a duplicate becomes a wrong number. They also matter to platform teams that operate the brokers those processors depend on.

How the idempotent producer deduplicates writes

When idempotence is enabled, the broker assigns the producer a producer ID (PID) at initialization. The producer then attaches a sequence number to every batch it sends, and that number is tracked separately for each partition. The broker stores the most recent sequence numbers it has accepted for each PID and partition pair, and it checks every incoming batch against them.

A retried batch whose sequence the broker has already written is a duplicate. The broker acknowledges it again and does not append it a second time. A batch whose sequence skips ahead means something earlier went missing, so the broker rejects it with an out-of-order sequence error rather than writing a gap.

Why acks=all is required: the deduplication state is rebuilt from the replicated log. Suppose a leader acknowledges a batch before followers have it and then fails. The new leader has no record of that sequence, so the guarantee does not hold across failover.

Multiple in-flight requests: these stay safe because the broker remembers a small window of recent batches per partition. Sequence checks reject any reordering that retries would otherwise cause. This is why idempotence limits how many requests can be in flight on a connection.

Limits:

• Deduplication is scoped to one producer session and one partition. A restarted producer gets a new PID unless it has a transactional.id, so its retries look like new data.
• Idempotence cannot make writes to several partitions atomic.
• It does nothing about consumers that reprocess records after a crash.
• Brokers also expire a producer's state after a period of inactivity. A producer that goes quiet for long enough loses its deduplication history.

Copy# producer.properties: idempotent producer, no transactions
bootstrap.servers=broker-1.example.internal:9092,broker-2.example.internal:9092
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

# Ask the broker for a producer ID (PID) and attach per-partition
# sequence numbers to every batch so retries can be deduplicated.
enable.idempotence=true

# Wait for all in-sync replicas. A leader that acks early and then fails
# leaves the new leader with no record of the sequence, which breaks dedup.
acks=all

# Retries are safe now: a resent batch with an already-written sequence
# is acknowledged again but not appended twice.
retries=2147483647

# Pipelining is allowed, but idempotence requires this to be 5 or fewer.
# The broker tracks a small window of recent batches per partition and
# uses it to reject reordering caused by retries.
max.in.flight.requests.per.connection=5

# No transactional.id is set, so the PID lasts only for this producer
# session. After a restart, a resent record gets a new PID and the broker
# treats it as new data. This config also cannot make writes to several
# partitions atomic, and it does not stop consumers from reprocessing records.

A Java client producer config that turns on idempotence with acks=all and a bounded number of in-flight requests, with comments explaining what the per-session, per-partition deduplication does and does not cover.

How Kafka transactions work: coordinator, transaction log and markers

Transactions add a transactional.id: a stable name you assign to a producer instance. It identifies the same logical producer across restarts.

Coordinator and transaction log

Kafka hashes the transactional.id to a partition of the internal __transaction_state topic. The broker that leads that partition is the producer's transaction coordinator. The coordinator records every state change of the transaction in that partition, so a newly elected coordinator can resume an unfinished transaction after a failover.

Copy            Producer (transactional.id = "billing-1")
              |  1. initTransactions -> PID + epoch
              |  2. AddPartitionsToTxn / AddOffsetsToTxn
              |  3. EndTxn (commit or abort)
              v
     Transaction coordinator ---> __transaction_state
              |                    (Ongoing, PrepareCommit, CompleteCommit)
              |  4. write markers
      +-------+--------------+-------------------+
      v                      v                   v
  orders-0              orders-3          __consumer_offsets-N
  [data..][COMMIT]      [data..][COMMIT]  [offsets..][COMMIT]

Lifecycle

• initTransactions gets a producer ID and increments its epoch. Any older instance that uses the same transactional.id is now fenced and its writes are rejected. The coordinator also completes or aborts any transaction the previous instance left open.
• beginTransaction only changes local producer state. It sends nothing to the broker.
• send writes records directly to partition leaders. Before the first write to a new partition, the producer registers that partition with the coordinator.
• sendOffsetsToTransaction adds the consumer group's __consumer_offsets partition to the transaction. The offsets are then written to that partition, but they stay uncommitted until the transaction ends.
• commitTransaction / abortTransaction:
  1. The coordinator logs PrepareCommit or PrepareAbort. After this point the outcome cannot change.
  2. It writes a control marker to every registered partition, including the offsets partition.
  3. It logs the completed state.

The output records and the consumed offsets get the same marker, so they become visible together or are discarded together.

Copy+-------------------------------+
| Producer                      |   3. Produce records (PID, epoch, seq)
| transactional.id = billing-1  |-----------------------------------+
+-------------------------------+   4. TxnOffsetCommit (via group   |
  |                                    coordinator)                 |
  | 1. InitProducerId -> PID, epoch                                 |
  | 2. AddPartitionsToTxn / AddOffsetsToTxn                         |
  | 5. EndTxn (commit | abort)                                      |
  v                                                                 |
+-------------------------------+     +---------------------+       |
| Transaction coordinator       |---->| __transaction_state |       |
| (leader of partition for      |     | Ongoing ->          |       |
|  hash(transactional.id))      |     | PrepareCommit ->    |       |
+-------------------------------+     | CompleteCommit      |       |
  |                                   +---------------------+       |
  | 6. WriteTxnMarkers                                              |
  +---------------------+-------------------+                       |
  |                     |                   |                       |
  v                     v                   v                       |
+--------------+  +--------------+  +------------------------+      |
| orders-0     |  | orders-3     |  | __consumer_offsets-N   |      |
| data, data,  |  | data, data,  |  | offsets,               |      |
| COMMIT       |  | COMMIT       |  | COMMIT                 |      |
+--------------+  +--------------+  +------------------------+      |
       ^                 ^                      ^                   |
       +-----------------+----------------------+-------------------+

The producer registers partitions and offsets with its transaction coordinator, writes data directly to partition leaders, then ends the transaction; the coordinator logs state to __transaction_state and writes COMMIT or ABORT markers to every involved partition, including __consumer_offsets.

Implementing a transactional read-process-write loop

The loop is short. Create the producer with a transactional.id, call initTransactions() once, and disable auto-commit on the consumer. For each poll: beginTransaction(), transform and send() the output records, then call sendOffsetsToTransaction(offsets, consumer.groupMetadata()) with the next offset to read for each input partition, and finally commitTransaction(). The offsets are written through the producer, not the consumer, which is why they share the transaction's marker.

initTransactions() registers the transactional.id with the coordinator and bumps its producer epoch. Any older instance still holding the same ID now has a stale epoch. The coordinator rejects its writes and commits, so a paused or partitioned "zombie" cannot complete a transaction after its replacement starts. This only works if the ID is stable across restarts of the same logical producer, such as a fixed ID per task or per deployment slot.

Passing the consumer group metadata adds a second check. The group coordinator rejects offset commits from a stale generation. As a result, an instance that lost partitions in a rebalance cannot commit offsets for them, even under a different transactional.id. Before this variant, the usual way to get that protection was one transactional.id per input partition.

The two common mistakes fail in opposite ways:

• Random IDs per start: No epoch is ever bumped against the old instance, so zombie fencing disappears.
• One ID shared by concurrent instances: Each initTransactions() fences the other, which produces constant fencing errors.

Treat ProducerFencedException as fatal for that instance: close the producer and stop. For other recoverable errors, call abortTransaction(), seek the consumer back to its last committed offsets, and retry the batch.

Copy// Assumes: producer has transactional.id="enricher-task-3" (stable per logical task),
// consumer has enable.auto.commit=false and isolation.level=read_committed.
producer.initTransactions(); // bumps the epoch and fences any older holder of this ID
consumer.subscribe(List.of("orders"));
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    if (records.isEmpty()) continue;
    producer.beginTransaction();
    try {
      Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
      for (ConsumerRecord<String, String> r : records) {
        producer.send(new ProducerRecord<>("orders-enriched", r.key(), r.value().toUpperCase()));
        // Commit the NEXT offset to read, not the offset just processed
        offsets.put(new TopicPartition(r.topic(), r.partition()),
                    new OffsetAndMetadata(r.offset() + 1));
      }
      // Group metadata lets the group coordinator reject commits from a stale generation
      producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
      producer.commitTransaction();
    } catch (ProducerFencedException e) {
      throw e; // A newer instance owns this transactional.id: stop, do not abort or retry
    } catch (KafkaException e) {
      producer.abortTransaction(); // Discard outputs and offsets from this batch
      // Rewind so the aborted input is processed again in the next transaction
      Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(consumer.assignment());
      for (TopicPartition tp : consumer.assignment()) {
        OffsetAndMetadata om = committed.get(tp);
        if (om != null) consumer.seek(tp, om.offset());
        else consumer.seekToBeginning(List.of(tp)); // or follow auto.offset.reset policy
      }
    }
  }
} finally {
  producer.close();
  consumer.close();
}

A Java consume-transform-produce loop. It commits input offsets inside the transaction with sendOffsetsToTransaction and consumer.groupMetadata(), treats ProducerFencedException as fatal, and on other errors aborts and rewinds to the last committed offsets.

Read committed consumers and the last stable offset

Downstream consumers do not see this protection unless they request it. The consumer setting isolation.level decides what a fetch can return, and its default is read_uncommitted.

The last stable offset is the offset of the first record that belongs to a still-open transaction. Everything below it is decided. A read_committed consumer never reads past it, so it cannot return a record whose outcome is unknown. Aborted data still sits in the log. The broker sends the consumer a list of aborted transactions that overlap the fetched range, and the client drops those records and skips the control markers.

The LSO is a property of the partition, not of a single producer. One long-running or hung transaction pins it. Every read_committed consumer on that partition stalls, including consumers that only want records from unrelated producers. The stall lasts until the transaction commits, aborts, or exceeds the producer's transaction.timeout.ms and is aborted by the coordinator. Keep that timeout tight and keep transactions short.

A minimal consumer setup is isolation.level=read_committed with enable.auto.commit=false. Apply it to every consumer. A single consumer left on the default will read aborted records with no error, and the exactly-once guarantee is gone for any system that consumer feeds.

Copy# consumer.properties: read only committed transactional data
bootstrap.servers=broker1:9092,broker2:9092
group.id=orders-projector

# The default is read_uncommitted, which returns records from open and aborted
# transactions. Set this explicitly on every consumer in the pipeline.
isolation.level=read_committed

# Commit offsets yourself after processing (or via the producer's
# sendOffsetsToTransaction in a consume-transform-produce loop),
# not on a background timer.
enable.auto.commit=false

# Only applies when the group has no committed offset yet.
auto.offset.reset=earliest

key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

# --- Producer side (separate file): bounds how long a hung transaction can
# --- pin the LSO and stall every read_committed consumer on the partition.
# transactional.id=orders-projector-0
# transaction.timeout.ms=60000

A consumer configured with isolation.level=read_committed and manual offset commits, plus the producer transaction.timeout.ms setting that limits how long one open transaction can hold back the last stable offset.

Where the guarantee ends and the alternatives beyond Kafka

Kafka transactions only cover Kafka. Output records and consumer offsets commit or abort together, but anything else the loop touches is outside that boundary. A database insert, an HTTP call or an email sent during processing has already happened by the time the transaction aborts or the producer is fenced. When the batch is retried, it happens again. That code must be idempotent on its own.

Three patterns extend the guarantee:

• Idempotent sinks. Derive a deterministic ID from the input, such as topic-partition-offset or a business key. Write with an upsert or a unique constraint so a replay overwrites the earlier write instead of duplicating it.
• Offsets stored in the sink. Write results and the consumed offset in one database transaction. On startup, seek to the stored offset and ignore Kafka's committed offsets.
• Outbox on the way in. A service writes its state change and an outbox row in one local transaction, and a relay publishes that row to Kafka. This removes the dual write at the source.

For topic-to-topic processing, Kafka Streams with processing.guarantee set to exactly-once is usually the better choice over a hand-rolled loop. It handles producer fencing, offset commits and state-store changelogs inside the transaction for you.

Avoid EOS when:

• The sink is external. It gives you nothing there.
• The pipeline is simple. Idempotent writes already make it correct.
• The team can't support it. EOS needs people who can diagnose hung transactions that block read-committed consumers.

Impact on cost, engineering time and operations

Every committed transaction pays a fixed cost no matter how many records it carries. The producer makes round trips to register partitions and consumer offsets with the coordinator. The coordinator writes to its transaction log and then writes a commit marker to each partition involved. That fixed cost shrinks as transactions grow, but a read_committed consumer sees none of a transaction's records until it commits.

Illustrative estimate (assumed numbers, not a benchmark): overhead fraction = F / (T + F), where F is the fixed commit cost and T is the processing time per transaction. Assume F = 10 ms. With T = 100 ms, overhead is about 9% and visibility delay is about 110 ms. With T = 1,000 ms, overhead drops to about 1%, but delay rises to about 1 s.

Operational checklist

• Set transaction.timeout.ms above your worst-case processing time. It must also stay below the broker's transaction.max.timeout.ms.
• Alert on transactions that stay open too long and on growing LSO lag, since one hung transaction blocks every read_committed reader on that partition.
• Treat ProducerFencedException as fatal: close the producer and do not retry.
• Confirm every downstream consumer sets isolation.level=read_committed, because the default read_uncommitted exposes aborted data.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to kafka exactly once semantics, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Changing a shared variable is one edit and one review instead of editing every service's .env by hand.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• A wrong or stale value is caught at review time by the services it affects, not by a failed deploy.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.