Idempotent Producers and Exactly-Once Semantics

In distributed streaming, the gap between at-least-once and exactly-once delivery is where production incidents breed. A network blip during a producer retry can duplicate a funds transfer; a consumer that crashes after producing output but before committing offsets can replay that output on restart. Apache Kafka’s idempotent producer and transactional exactly-once semantics (EOS) are the primary mechanisms that close this gap. This article dissects how these features work, how to configure them correctly, and how to avoid the operational pitfalls that quietly undermine their guarantees. We build on Producers, Consumers & Delivery Semantics, extending it into the realm of strict correctness. For the consumer-side complement, Offset Management Strategies for Reliable Processing covers the commit strategies that pair with the producer guarantees discussed here.

The Problem: Duplicates and Reordering on Retry

Kafka producers operate in an adversarial environment: networks drop packets, brokers fail mid-append, and leader elections cause metadata staleness. The producer’s defense is the retry. If a produce request fails with a retriable error—NOT_LEADER_OR_FOLLOWER, NETWORK_EXCEPTION, or a request timeout—the producer resends the batch. That retry is also the root cause of duplication.

Consider a producer sending with acks=all. The batch reaches the leader, is written to the log, and replicates to the in-sync replicas. The leader then sends the acknowledgment. If that acknowledgment is lost, the producer times out and retries; without deduplication, the leader appends the same batch again and the consumer sees two identical records. This is the classic at-least-once duplicate.

The problem compounds with max.in.flight.requests.per.connection > 1. If the producer pipelines requests—sending batch B1, then B2 before B1 is acknowledged—and B1 fails and is retried after B2 succeeds, the log ordering becomes B2, B1. For applications that rely on per-partition ordering, this reordering is as damaging as duplication.

The idempotent producer addresses both at the broker, by tagging each batch with a producer ID and sequence number so the broker can detect duplicates and reject out-of-order batches while preserving order. The alternative—pushing deduplication into every consumer, backed by an external state store—is far more expensive and error-prone.

How the Idempotent Producer Works: PIDs and Sequence Numbers

Idempotence is not a separate producer type but a mode enabled with enable.idempotence=true. Since Kafka 3.0 (KIP-679) it is on by default: if you set no conflicting configs, you get an idempotent producer automatically. On its first request the producer performs an InitProducerId handshake, and the broker assigns it a Producer ID (PID), a 64-bit integer stable for the lifetime of that producer session. The producer then tags every batch to a given topic-partition with a monotonically increasing sequence number starting at 0.

For each active producer session the broker tracks (PID, topic-partition) → last sequence. When a produce request arrives, the broker validates the sequence:

• If it is exactly last + 1, the batch is appended and the sequence advances.
• If it is less than or equal to last, the batch is a duplicate and is discarded, but the broker still returns success. This is the crucial step: the producer sees success and stops retrying, breaking the duplication loop.
• If it is greater than last + 1, the broker detects a gap and returns OutOfOrderSequenceException. The producer cannot fill the gap (it has already moved on), so this is a fatal error.

To preserve ordering, the idempotent producer requires max.in.flight.requests.per.connection <= 5. Within that bound the producer guarantees in-order delivery even across retries, because the broker can buffer and reorder up to five in-flight batches by sequence number. The default of 5 is almost always correct; dropping to 1 disables pipelining and can sharply reduce throughput.

PID/sequence state is in-memory and is lost on a hard broker failure of the partition leader. This does not break idempotency: a failover triggers a leader change, the producer sees NOT_LEADER_OR_FOLLOWER, refreshes metadata, and continues with the same PID against the new leader, whose log already reflects the replicated sequence state. The mechanism is designed so that the producer never has to reason about which leader holds which state.

Configuring the Idempotent Producer

# Idempotent delivery (these are the 3.0+ defaults; shown explicitly)
enable.idempotence=true
acks=all
retries=2147483647
max.in.flight.requests.per.connection=5

When enable.idempotence=true, Kafka enforces three requirements:

• acks=all (i.e. -1). Idempotence with acks=1 is unsafe: a new leader after failover may not hold the batch, so the producer’s retry would be treated as a genuinely new message.
• retries defaults to Integer.MAX_VALUE (2147483647). The producer should not abandon a retry, because a retry that the broker actually committed would otherwise be lost.
• max.in.flight.requests.per.connection <= 5.

If you explicitly set a conflicting value (for example acks=1) together with enable.idempotence=true, the client throws a ConfigException at construction. If idempotence is left at its default and you set a conflicting value, the client silently falls back to a non-idempotent producer—so check your effective config rather than assuming EOS is active.

Session Lifetime and PID Expiry

A producer session is tied to the producer instance. If the process crashes and a new instance starts, it receives a new PID; batches the old instance sent and the new instance re-sends are not recognized as duplicates. Keep producer instances long-lived, and in Kubernetes avoid aggressive memory limits that trigger OOM kills mid-flight.

On the broker, idempotent-producer PID state is evicted by the producer-ID expiration mechanism (governed by producer.id.expiration.ms, default 1 day) once a PID has been idle. This is distinct from transactional.id.expiration.ms (default 7 days), which governs how long the coordinator retains state for a transactional transactional.id. The distinction matters when debugging: a non-transactional idempotent PID and a transactional ID age out under different settings.

Transactions: Atomicity Across Partitions and Topics

The idempotent producer guarantees no duplicates and no reordering within a single topic-partition. It does not provide atomicity across partitions or topics: a producer writing to both a payments and an audit topic can have one write land and the other fail. Kafka transactions group produce operations into atomic units that either fully commit or fully abort.

Transactions are built on the idempotent producer—enable.idempotence=true is implied and forced. The producer is configured with a stable transactional.id that survives restarts. Unlike the ephemeral PID, the transactional.id lets the coordinator associate a new producer instance with prior state and fence zombie producers: older instances believed dead but still attempting to write.

The protocol adds two pieces: the Transaction Coordinator, a broker component that owns transaction state in the internal __transaction_state topic, and transaction markers (commit/abort control records) written to each involved partition. Consumers configured with isolation.level=read_committed see only records from committed transactions, filtering aborted ones and holding back records from still-open transactions.

Configuring a Transactional Producer

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("transactional.id", "payment-service-txn-01");
props.put("enable.idempotence", "true"); // implied by transactional.id, explicit is fine
props.put("acks", "all");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("payments", key, paymentJson));
    producer.send(new ProducerRecord<>("audit", key, auditJson));
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // Fatal: the producer is defunct. Close and let orchestration replace it.
    producer.close();
    throw e;
} catch (KafkaException e) {
    // Abortable: roll back and decide whether to retry the whole transaction.
    producer.abortTransaction();
}

The catch clauses match the pattern in the KafkaProducer Javadoc: ProducerFencedException, OutOfOrderSequenceException, UnsupportedVersionException, and AuthorizationException are fatal and leave only close(); other KafkaExceptions are abortable. The transactional.id must be unique per logical producer; if two instances share one ID, the newer instance fences the older, and the older gets a ProducerFencedException on its next call. This fencing is the basis of zombie detection during failover.

The Transaction Coordinator and Two-Phase Commit

Kafka’s transaction protocol is a tailored two-phase commit. The producer first sends data records to the relevant partitions; they are written to the log but withheld from read_committed consumers. On commit, the coordinator writes a PrepareCommit record to __transaction_state, writes commit markers to every involved partition, then writes a CompleteCommit record. If any step fails, the coordinator writes abort markers instead.

This has direct operational consequences. __transaction_state is a compacted internal topic created with transaction.state.log.replication.factor (default 3) and transaction.state.log.min.isr (default 2; effective only because the replication factor is 3). If the coordinator cannot make progress writing to __transaction_state—because it is under-replicated or its leaders are unstable—transactions stall. Treat the under-replication and leader stability of this topic as a first-class alert.

For atomic multi-topic writes and offset-commit integration in depth, see Using Transactions to Atomically Produce to Multiple Topics.

Consumer-Side EOS: Isolation Levels and Offset Integration

Producer guarantees are only half the story. The consumer participates through isolation.level:

• read_uncommitted (default): returns all records, including those from open or aborted transactions. Lowest latency; acceptable only where rolled-back records can be tolerated.
• read_committed: returns records only from committed transactions; aborted records are filtered and open-transaction records are withheld until the transaction resolves. This adds latency bounded by transaction duration.

With read_committed, the consumer cannot advance past the Last Stable Offset (LSO)—the offset of the earliest still-open transaction. A producer that hangs with an open transaction therefore stalls every read_committed consumer of those partitions until the transaction commits, aborts, or times out. Note that max.poll.records bounds records returned per poll(), not internal buffering. Keep transactions short (ideally seconds) and rely on the broker timeout as a backstop.

Integrating Offset Commits with Transactions

For end-to-end EOS in a consume-transform-produce loop, the consumer’s offset commit must be part of the same transaction as the output. sendOffsetsToTransaction does exactly this. Note there is no positionToOffsets() helper on KafkaConsumer; you build the offsets map yourself from consumer.position() per assigned TopicPartition:

producer.beginTransaction();
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
    producer.send(new ProducerRecord<>("output-topic", record.key(), transform(record.value())));
}

// Build the offsets-to-commit map: next offset to read = last processed + 1.
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
for (TopicPartition partition : records.partitions()) {
    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
    offsets.put(partition, new OffsetAndMetadata(lastOffset + 1));
}

producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();

The signature is sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata>, ConsumerGroupMetadata). Passing consumer.groupMetadata() (rather than just the group ID) is what enables the per-instance fencing introduced in KIP-447, so a zombie consumer-producer cannot commit stale offsets after a rebalance. If the transaction commits, output and offset advance together durably; if it aborts, neither is visible to read_committed consumers and the input is reprocessed. This is the core of EOS in Kafka Streams.

During a rebalance a consumer can lose partitions mid-loop. The producer must abort the in-progress transaction, and the application must handle the CommitFailedException that arises if it tries to commit offsets for partitions it no longer owns. See Consumer Group Rebalancing: Protocols and Tuning.

Operational Tuning and Performance

Idempotence and transactions are not free; account for the overhead in capacity planning.

Idempotent Producer Overhead

The per-batch cost of idempotence—a hash lookup and an integer comparison on the broker—is on the order of microseconds and is dominated entirely by the forced acks=all, which adds the replication round-trip to produce latency. In a multi-region ISR this can add tens of milliseconds. Keep the ISR co-located in a low-latency zone and use replication factor 3 with min.insync.replicas=2 for a sound durability/latency balance.

Transactional Producer Overhead

Each beginTransaction/commitTransaction adds coordinator round-trips, and each commit writes a marker to every partition touched—one extra control batch per partition. Transactions that fan out over many partitions pay proportionally. You can benchmark the difference with kafka-producer-perf-test:

# Idempotent (non-transactional) baseline
kafka-producer-perf-test \
  --topic test-topic --num-records 10000000 --record-size 1024 --throughput -1 \
  --producer-props bootstrap.servers=broker1:9092,broker2:9092 \
    enable.idempotence=true acks=all

# Transactional run (set transaction.timeout.ms within the broker max)
kafka-producer-perf-test \
  --topic test-topic --num-records 10000000 --record-size 1024 --throughput -1 \
  --producer-props bootstrap.servers=broker1:9092,broker2:9092 \
    enable.idempotence=true acks=all transactional.id=perf-test-01 \
    transaction.timeout.ms=60000

Note that kafka-producer-perf-test issues beginTransaction/commitTransaction only when --transactional-id is supplied as a dedicated CLI flag; passing transactional.id via --producer-props alone configures the producer but, depending on version, may not drive the transactional loop. Check kafka-producer-perf-test --help for your release. Empirically, transactional throughput runs roughly 10–30% below the idempotent baseline depending on transaction size; batching more records per transaction amortizes the commit cost. Committing every 100–1000 records or every ~100 ms (whichever comes first) keeps transactions short while preserving throughput.

Broker-Side Tuning for Transactions

transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
transaction.state.log.segment.bytes=104857600
transaction.max.timeout.ms=900000
transaction.abort.timed.out.transaction.cleanup.interval.ms=60000

transaction.max.timeout.ms (default 900000 ms = 15 minutes) caps how long any transaction may stay open. A producer may request a shorter transaction.timeout.ms but not a longer one; requesting longer fails with InvalidTxnTimeoutException. Setting the broker max too high lets a stuck producer hold the LSO and stall read_committed consumers; too low risks aborting legitimate long transactions.

transaction.abort.timed.out.transaction.cleanup.interval.ms (default 60000 ms) controls how often the broker scans for and aborts timed-out transactions, so a transaction can linger up to roughly transaction.timeout.ms + 60s before cleanup. The scan is cheap, but a rising rate of timed-out transactions is a leading indicator of producer trouble.

Failure Modes and Recovery

Producer Fencing and Zombies

Fencing is the most common transactional failure. When a newer instance with the same transactional.id calls initTransactions(), the coordinator bumps the producer epoch; the older instance receives ProducerFencedException on its next call. This is fatal—close the producer and let orchestration replace it. The epoch bump is precisely what defeats a network-partitioned zombie: its writes carry the stale epoch and are rejected.

Handling OutOfOrderSequenceException

This is thrown when the broker sees a sequence gap for a (PID, topic-partition) it cannot reconcile. The producer cannot retransmit the missing sequences, so the error is fatal: close and recreate the producer rather than retrying the batch. The forced retries=MAX_VALUE and a stable network minimize the chance of hitting it; in cloud environments, avoid load-balancer idle timeouts and network policies that silently drop long-lived broker connections.

Transaction Timeouts and Crash Recovery

When a transaction times out, the broker writes abort markers and the producer’s next commitTransaction fails (typically with a timeout/KafkaException). Call abortTransaction() to clear local state, then decide whether to replay. Because replay re-consumes input, transactional processing should be designed so reprocessing the same input yields the same output.

A subtle case: the producer crashes after sending records but before committing. The transaction stays open—blocking read_committed consumers behind the LSO—until it times out or the producer restarts. On restart, initTransactions() for the same transactional.id bumps the epoch and aborts any in-progress transaction from the prior epoch, restoring a clean state.

Offset Desynchronization

With sendOffsetsToTransaction, a failed commit leaves offsets unadvanced, so the consumer re-reads on restart. This only yields EOS if the downstream consumer reads read_committed; a common mistake is making the producer transactional but leaving the downstream consumer at the default read_uncommitted, which exposes records from aborted transactions and reintroduces double-processing. Always pair transactional producers with read_committed consumers.

Monitoring Exactly-Once Pipelines

EOS is only as trustworthy as the monitoring that validates it.

Producer Metrics

The JmxTool class moved to the org.apache.kafka.tools package (KIP-906); the old kafka.tools.JmxTool path was deprecated in 3.5 and removed in 4.0. Use the new class name (and the kafka-jmx.sh wrapper where shipped):

# Poll producer client metrics over JMX (Kafka 3.5+/4.x)
kafka-run-class org.apache.kafka.tools.JmxTool \
  --object-name 'kafka.producer:type=producer-metrics,client-id=*' \
  --attributes record-error-rate,record-retry-rate,record-send-rate,outgoing-byte-rate,request-latency-avg \
  --jmx-url service:jmx:rmi:///jndi/rmi://producer-host:9999/jmxrmi

Producer metrics are exposed by the client, so point JMX at the producer JVM, not a broker. Alert on:

• record-error-rate — sustained non-zero values indicate network, broker, or config problems. Deployment-correlated spikes are expected; sustained errors are not.
• record-retry-rate — elevated retries signal network instability or overloaded brokers; with idempotent producers retrying indefinitely, high retry rates translate directly into latency.

For transactional cost, the producer exposes cumulative timing counters—txn-commit-time-ns-total, txn-abort-time-ns-total, txn-begin-time-ns-total, txn-init-time-ns-total, and txn-send-offsets-time-ns-total—on producer-metrics. A rising txn-abort-time-ns-total relative to txn-commit-time-ns-total flags transactions that are aborting rather than committing. (There is no transaction-abort-rate/transaction-commit-rate attribute on the producer; rate it yourself from these -total counters.)

Broker-Side Transaction Coordinator Metrics

The broker exposes transaction-coordinator load timing under kafka.server:type=transaction-coordinator-metrics:

kafka-run-class org.apache.kafka.tools.JmxTool \
  --object-name 'kafka.server:type=transaction-coordinator-metrics,*' \
  --attributes partition-load-time-max,partition-load-time-avg \
  --jmx-url service:jmx:rmi:///jndi/rmi://broker1:9999/jmxrmi

partition-load-time-max/partition-load-time-avg report how long it took to load transaction-state metadata from __transaction_state (measured over the last 30s); spikes indicate coordinator failover or a slow/under-replicated state topic. Apache Kafka does not publish a documented active-transaction-count or transaction-expired-count attribute here, so do not build alerts on those names—monitor open-transaction risk indirectly via consumer lag relative to the LSO and the rate of broker-side transaction aborts in the logs.

Lag for read_committed Consumers

For read_committed consumers, records-lag-max is measured against the LSO, not the log end offset, so lag can appear elevated simply because an upstream transaction is still open—not because the consumer is slow. Supplement lag with end-to-end latency: inject heartbeat records into the input at a fixed interval and measure their propagation through the pipeline. That captures transaction-withholding, processing time, and queuing in a single number.

Integration with Stream Processing

Most production EOS pipelines run on frameworks that hide the low-level APIs. Kafka Streams enables EOS with processing.guarantee=exactly_once_v2. exactly_once_v2 (KIP-447) is the current implementation; the original exactly_once is deprecated and the very old exactly_once_beta alias was removed. Internally, Streams wires its producers and consumers to use transactions with read_committed and the sendOffsetsToTransaction pattern above, handling fencing, timeouts, and rebalances for you.

When running Streams with EOS, watch transaction.timeout.ms. The Streams default is 10000 ms (10s) historically and was raised to a larger value in recent releases—verify the default for your version—and it may be too short for stateful operations that make remote calls. Raise it in proportion to worst-case per-batch processing time, but keep it under the broker’s transaction.max.timeout.ms. For a full walkthrough including Spring Kafka, see Achieving Exactly-Once Semantics with Kafka Streams and Spring Kafka.

Security: ACLs for Transactional Producers

In an ACL-secured cluster, a transactional producer’s principal needs Write and Describe on the TransactionalId resource (the Describe check happens when the producer looks up its coordinator), plus the usual Write on the destination topics. Producers do not need ACLs on the internal __transaction_state topic—that topic is accessed by the coordinator on the broker, not by client principals.

# TransactionalId permissions for the producer principal
kafka-acls --bootstrap-server broker1:9092 --command-config admin-client.conf \
  --add --allow-principal User:payment-service \
  --operation Write --operation Describe \
  --transactional-id payment-service-txn-01

# Write access to the destination topics it produces to
kafka-acls --bootstrap-server broker1:9092 --command-config admin-client.conf \
  --add --allow-principal User:payment-service \
  --operation Write --operation Describe \
  --topic payments --topic audit

Granting Write on the TransactionalId also implicitly authorizes idempotent writes for that producer. Without these grants, initTransactions() fails with a TransactionalIdAuthorizationException. Treat the transactional.id as sensitive: anyone who can use a given ID can fence the legitimate producer and cause a denial of service, so namespace IDs per tenant and scope ACLs tightly in multi-tenant clusters.

Capacity Planning

EOS raises resource demand along several axes. Forced acks=all increases inter-broker replication traffic; transaction markers add one control batch per partition per commit/abort; and __transaction_state sees write amplification proportional to transaction count. As a rule of thumb, budget roughly a 20–30% increase in broker CPU and network for the same message volume when moving from at-least-once to EOS, with the acks=all replication traffic dominating.

Marker overhead itself is small. With markers on the order of a few dozen bytes per partition plus a __transaction_state write per commit/abort, a 10-partition transaction committed every 100 ms adds only single-digit KB/s—negligible beside replication.

On consumers, read_committed holds back data behind the LSO; in the worst case (a long-running upstream transaction) a consumer may need to buffer the data produced during that transaction’s lifetime. For a 10 MB/s producer with a 5-second transaction, that is on the order of 50 MB. Size consumer heaps with this in mind and keep GC pauses well under the session timeout.

Conclusion

The idempotent producer and transactional EOS turn Kafka from a high-throughput bus into a platform fit to serve as a system of record. Their guarantees—no duplicates, no per-partition reordering, and atomic multi-partition writes plus offset commits—are the foundation for correctness-critical systems. But none of it is automatic. The idempotent producer must run with acks=all and effectively unbounded retries (the 3.0+ defaults). Transactions need stable transactional.id values, short transaction durations, and read_committed consumers downstream. Verify your effective configuration rather than assuming, monitor the LSO and __transaction_state health, and the operational cost stays well below the alternative of building deduplication and atomicity in application code.

Start from Producers, Consumers & Delivery Semantics and the consumer-side strategies in Offset Management Strategies for Reliable Processing. Kafka provides the primitives; wielding them with precision is the operator’s job.