Exactly-Once Processing in Kafka Streams

Exactly-once semantics (EOS) means each input record affects the application’s state and output exactly once, even when producers retry, brokers fail, or consumers rebalance. Kafka Streams delivers this by wrapping the consume-transform-produce cycle in a Kafka transaction, binding state-store updates, output writes, and source-offset commits into one atomic unit. This article covers the transactional mechanics, the configuration that actually matters in production, failure recovery and zombie fencing, the real performance cost, and the metrics and tools you need to operate EOS pipelines.

At-Most-Once, At-Least-Once, and Exactly-Once

The three delivery semantics that govern Stream Processing outcomes differ in how they handle the gap between processing a record and committing its offset. At-most-once commits offsets before processing, so records can be lost on a crash but are never reprocessed. At-least-once commits offsets after processing, guaranteeing no loss but allowing duplicates when a failure occurs between the output write and the offset commit. Exactly-once eliminates both loss and duplication.

Kafka Streams builds EOS on top of at-least-once by adding a transactional boundary around each commit cycle. When a stream thread commits, it writes pending output records, flushes state-store changelog records, and commits the source offsets as one transaction. Either all of these become visible to downstream consumers reading with isolation.level=read_committed, or none of them do.

The trade-off is concrete: at-least-once pushes deduplication onto downstream consumers, while EOS moves that burden into the Streams runtime at the cost of latency and throughput. Decide deliberately rather than enabling EOS by default:

The choice between the Kafka Streams DSL vs Processor API: When to Use Each does not change this: both inherit transactional behavior from processing.guarantee, so you never implement transactions manually in the Processor API.

Transactional Mechanics in Kafka Streams

Set processing.guarantee=exactly_once_v2 to enable EOS. The framework derives a transactional ID for each producer from the application ID, and—under EOS v2—a single thread-level producer handles all tasks assigned to that thread, so the ID is stable across restarts and rebalances.

Before a transaction can begin, the producer registers its transactional ID with the transaction coordinator, a broker-side component that manages transaction lifecycles. The coordinator persists transaction metadata to the internal __transaction_state topic, the durable source of truth for transaction status. On each commit, Streams writes output records, sends the consumer offsets to the transaction via sendOffsetsToTransaction, and commits. Under read_committed, those records stay invisible until the coordinator writes commit markers to the affected partitions.

exactly_once_v2 (KIP-447) improves on the original exactly_once (now removed) by using one producer per stream thread instead of one per input partition. The old model created a producer—and a distinct transactional ID and broker connection—for every input partition a thread owned, so a topology with hundreds of partitions opened hundreds of transactional sessions per instance. EOS v2 collapses that to one per thread by sending the consumer group metadata alongside the offsets, letting the broker fence on group identity instead of partition identity. This is what makes EOS scale to large partition counts. The EOS v2 protocol requires brokers running 2.5 or later.

// Kafka Streams configuration for exactly-once semantics v2
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "inventory-eos-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
// Producer transaction timeout, applied via the producer prefix
props.put(StreamsConfig.PRODUCER_PREFIX + ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60000);

Naming and versions. The EOS v2 protocol shipped in Kafka 2.5 under the config value exactly_once_beta. KIP-732 (Kafka 3.0) renamed it to exactly_once_v2, deprecated both exactly_once and exactly_once_beta, and those deprecated values—along with the old eos-alpha implementation—were removed in 4.0. Use the StreamsConfig.EXACTLY_ONCE_V2 constant rather than a string literal.

Transactional ID fencing prevents zombie instances from corrupting output. When a task moves to a new instance, the new producer initializes with the same transactional ID and the coordinator bumps the producer epoch. Any in-flight transaction from the old producer is then rejected with ProducerFencedException and aborted. This fencing is the core correctness mechanism in EOS—without it, a slow-to-die old instance could commit stale output after a new instance had already taken over.

Configuring Exactly-Once in Production

EOS configuration extends beyond processing.guarantee into broker durability settings and producer requirements.

At the broker level, set transaction.state.log.replication.factor to at least 3 and transaction.state.log.min.isr to at least 2 so transaction metadata survives broker failures and is acknowledged by a quorum before a commit is considered durable. These are cluster-wide and require a rolling restart. The broker defaults already are replication factor 3 and min ISR 2, which suits production—but a single-broker or small dev cluster will fail to create the __transaction_state topic if it cannot satisfy them, so lower them explicitly for local testing.

# Broker configuration in server.properties
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
transaction.abort.timed.out.transaction.cleanup.interval.ms=10000
transaction.remove.expired.transaction.cleanup.interval.ms=3600000

On the application side, transaction.timeout.ms bounds how long a transaction may stay open before the coordinator aborts it. The producer-side default is 60000 ms. Kafka Streams’ commit cycle is driven by commit.interval.ms, which defaults to 100 ms when EOS is enabled (versus 30000 ms otherwise), so transactions are normally short-lived. Raise transaction.timeout.ms comfortably above your worst-case commit cycle for windowed aggregations over large Stateful Stream Processing with Kafka Streams state. Crucially, the broker’s transaction.max.timeout.ms (default 900000 ms / 15 min) caps this value—requests above it are rejected, so a producer asking for a longer timeout fails at initTransactions() rather than silently getting the cap.

// Raise the producer transaction timeout for long commit cycles
props.put(StreamsConfig.PRODUCER_PREFIX + ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 120000);

You do not need to set enable.idempotence or acks manually: with processing.guarantee=exactly_once_v2, Kafka Streams forces enable.idempotence=true and acks=all on its internal producers and overrides conflicting values. Setting them is harmless but redundant.

Internal repartition and changelog topics should meet the same durability bar as your data. Override their replication factor with replication.factor, and apply topic-level overrides with StreamsConfig.topicPrefix(String), which prefixes a single TopicConfig property applied to all internal topics at creation time.

// Replication factor for internal (repartition/changelog) topics
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
// Apply a TopicConfig override to internal topics, e.g. segment sizing
props.put(StreamsConfig.topicPrefix(TopicConfig.SEGMENT_BYTES_CONFIG), "536870912");

topicPrefix applies to every internal topic, so it is not the place to set cleanup.policy: Streams already creates key-value changelog topics with cleanup.policy=compact and repartition topics with cleanup.policy=delete automatically. A global cleanup.policy override would misconfigure one or the other.

Failure Recovery and Zombie Fencing

When an instance fails, EOS guarantees no partial results are exposed. Recovery has three phases: fencing the old producer, aborting its open transaction, and rebuilding state.

The coordinator bounds open transactions by time, not heartbeats. If a producer goes silent and its open transaction exceeds transaction.timeout.ms, the coordinator aborts it, which unblocks read_committed consumers that would otherwise stall at the Last Stable Offset (LSO) behind that open transaction. The LSO is the key concept: a read_committed consumer can never read past the earliest still-open transaction on a partition, so one stuck transaction stalls every such consumer on that partition.

When a new instance takes over a task, its producer initializes with the same transactional ID, the coordinator bumps the epoch, and the old producer is fenced with ProducerFencedException. The new instance restores its state store by replaying the task’s changelog topic, then resumes from the last committed source offset.

# Describe the state of a transactional ID
kafka-transactions.sh --bootstrap-server broker1:9092 \
  describe --transactional-id inventory-eos-app-0_0

# Example output:
# CoordinatorId TransactionalId        ProducerId ProducerEpoch TransactionState TransactionTimeoutMs ...
# 1             inventory-eos-app-0_0  12345      3             Ongoing          60000               ...

The dangerous case is a hanging transaction: an open transaction whose producer is gone but which the coordinator never aborted (for example, after certain broker-side failures). It pins the LSO and silently blocks every read_committed consumer on that partition while contributing no obvious error—throughput simply flatlines. Detect and resolve these with the kafka-transactions.sh tooling from KIP-664 rather than guessing.

# Find transactions left open longer than the given timeout (ms),
# scoped to a topic or a broker
kafka-transactions.sh --bootstrap-server broker1:9092 \
  find-hanging --topic orders --max-transaction-timeout 900000

# Abort an identified hanging transaction on a partition
kafka-transactions.sh --bootstrap-server broker1:9092 \
  abort --topic orders --partition 3 --start-offset 1000

Aborting is safe by design: it discards uncommitted output that was never visible to read_committed consumers anyway. The risk is doing it to the wrong transaction, so always confirm with find-hanging first and abort only the exact partition and start-offset it reports.

Performance Overhead and Capacity Planning

EOS adds overhead in three places: extra coordinator round-trips per commit, the disk I/O of persisting transaction state, and a higher commit frequency. Because Streams defaults commit.interval.ms to 100 ms under EOS, the dominant cost is usually commit frequency, not per-record work—each commit involves an offset-send and a transaction commit to the coordinator plus commit-marker writes to every output partition.

The practical effect is added latency per commit and reduced throughput versus at-least-once. The exact numbers depend heavily on record size, partition count, cluster load, and network topology, so benchmark with your own workload rather than relying on a fixed percentage. Commit markers also add records to every output and changelog partition, so factor a modest write-amplification overhead into changelog and __transaction_state disk provisioning.

The most effective throughput lever is raising commit.interval.ms to amortize coordination over more records, accepting higher end-to-end latency in return—appropriate when SLAs are measured in seconds rather than milliseconds. Two second-order effects to keep in mind: a longer interval holds each transaction open longer, so it must stay well under transaction.timeout.ms; and it lets more uncommitted output accumulate, which delays visibility for read_committed consumers by roughly the interval.

// Trade latency for throughput by committing less frequently
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); // commit every 1s

// Tune internal consumer/producer via the correct prefixes
props.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), 1000);
props.put(StreamsConfig.producerPrefix(ProducerConfig.BATCH_SIZE_CONFIG), 32768);   // 32 KB batches
props.put(StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG), 10);       // wait 10ms to fill

Monitoring and Troubleshooting

Operating EOS means watching the transaction coordinator, the producers, and any read_committed consumers that could stall behind an open transaction.

The coordinator exposes the MBean kafka.server:type=transaction-coordinator-metrics. Its partition-load-time-avg and partition-load-time-max attributes report how long it took to load transaction state from __transaction_state partitions over the last 30 seconds; a spike there points to a coordinator failover or a slow disk. Watch the broker request metrics for WriteTxnMarkers and InitProducerId/AddPartitionsToTxn/EndTxn under kafka.network:type=RequestMetrics to gauge coordinator load.

On the producer side, the EOS-relevant attributes live under kafka.producer:type=producer-metrics,client-id=<id> and are cumulative nanosecond counters: txn-commit-time-ns-total, txn-abort-time-ns-total, txn-init-time-ns-total, txn-begin-time-ns-total, and txn-send-offsets-time-ns-total. In steady state, txn-commit-time-ns-total should climb smoothly while txn-abort-time-ns-total stays flat—a steadily climbing abort counter means transactions are timing out or being fenced and warrants investigation.

The fastest way to operate this is to map each observable symptom to its likely cause and the tool that confirms it:

# Fenced/zombie producers surface in the broker log
grep -i "ProducerFenced\|InvalidProducerEpoch" /var/log/kafka/server.log | tail -20

# A read_committed consumer stuck behind an open transaction shows a
# growing LAG that does not advance even while the producer is writing
kafka-consumer-groups.sh --bootstrap-server broker1:9092 \
  --group inventory-eos-app --describe

The official Kafka documentation on message delivery semantics describes the transactional protocol, and KIP-98 covers the original transactional design.

Operational Best Practices and Trade-Offs

Pin processing.guarantee in your deployment manifests and treat it as a fixed property of an application ID. Switching an existing application between at-least-once and EOS changes how offsets and transactions are managed, so reset the application’s state first with kafka-streams-application-reset.sh. Mind the flag names: input topics are reset with --input-topics, internal topics are deleted with --internal-topics (there is no --intermediate-topics flag), and --bootstrap-server is the preferred form.

# Reset application state before changing processing.guarantee
kafka-streams-application-reset.sh \
  --bootstrap-server broker1:9092 \
  --application-id inventory-eos-app \
  --input-topics source-topic \
  --internal-topics inventory-eos-app-KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition

Test failure paths deliberately. Inject network faults and broker partitions with a tool like Toxiproxy and confirm pipelines recover cleanly. Pay particular attention to the interaction between transaction.timeout.ms, max.poll.interval.ms, and rebalances: if a commit cycle stalls long enough to exceed max.poll.interval.ms, the consumer leaves the group, triggering a rebalance that fences the producer and aborts its transaction—a self-reinforcing loop if the underlying stall persists.

Finally, question whether you need EOS at all. Many pipelines are well served by at-least-once plus idempotent sinks—a downstream store that upserts by primary key naturally absorbs duplicates—and the operational simplicity often outweighs the marginal correctness benefit. When you do need EOS, run exactly_once_v2 on brokers 2.5+ and migrate off any remaining deprecated exactly_once deployments, which no longer exist as of Kafka 4.0.