Achieving Exactly-Once Semantics with Kafka Streams and Spring Kafka

Exactly-once semantics (EOS) means each input record is reflected exactly once in the application’s output and committed offsets, even across crashes and rebalances. This matters for financial postings, inventory adjustments, and any stateful consume-transform-produce pipeline where a duplicate or a dropped record is a correctness bug, not a metric blip. Kafka Streams and Spring Kafka both expose EOS, but it is a property of the whole pipeline and depends on a few exact settings being correct. This guide covers the underlying transactional machinery, the precise configuration for both frameworks, failure behavior, and the metrics that actually exist.

Prerequisites: familiarity with Producers, Consumers & Delivery Semantics and Idempotent Producers and Exactly-Once Semantics.

How Kafka Achieves Exactly-Once Semantics

Kafka’s EOS guarantee spans the consume-transform-produce cycle. Three mechanisms combine:

1. Idempotent producer. The producer receives a producer ID (PID) from the broker, and each batch carries a monotonically increasing sequence number per partition. The broker deduplicates on (PID, topic-partition, sequence), eliminating duplicates from producer retries. Enabled with enable.idempotence=true (the default since Kafka 3.0).

2. Transactional producer. Groups writes across multiple topic-partitions into an atomic unit via beginTransaction() / commitTransaction() / abortTransaction(). Crucially, the consumer offsets are written into the same transaction through sendOffsetsToTransaction(...), so the consumed position advances only when the produced output is committed. Requires a transactional.id.

3. Transaction coordinator. A broker-side component that persists transaction state to the internal __transaction_state topic (50 partitions by default, set by transaction.state.log.num.partitions). On commit it writes transaction markers to every partition the transaction touched. Consumers configured with isolation.level=read_committed skip records belonging to aborted or still-open transactions, returning only data up to the last stable offset (LSO).

Zombie fencing. Each transactional.id is associated with a producer epoch. When a new producer instance initializes with the same transactional.id, the coordinator bumps the epoch and fences the previous instance: requests from the stale epoch are rejected with ProducerFencedException. This is what prevents a hung or partitioned old instance from corrupting committed state.

Configuring EOS in Kafka Streams

Kafka Streams turns on EOS with a single setting: processing.guarantee. The default is at_least_once — EOS is not on unless you explicitly set it. Set processing.guarantee=exactly_once_v2 to enable the full transactional pipeline. The legacy value exactly_once (EOS v1, one producer per task) and the interim exactly_once_beta were deprecated and removed in Kafka 4.0; use exactly_once_v2, which requires brokers on 2.5 or newer.

exactly_once_v2 (introduced as exactly_once_beta in 2.6, renamed in 3.0) uses a single producer per StreamThread and folds offset commits into record production, sharply reducing the number of producers and coordinator round-trips compared to v1. It relies on epoch-based zombie fencing rather than a producer-per-task.

When EOS is enabled, Streams automatically configures its internal clients for you: producers get enable.idempotence=true and consumers get isolation.level=read_committed. You do not need to set those manually for the internal clients; set them only on any external KafkaConsumer that reads the output topics.

Required broker-side settings

Note that transaction.timeout.ms is set on the producer, but it cannot exceed the broker’s transaction.max.timeout.ms (15 minutes by default); the broker rejects a higher value.

Streams application configuration

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "inventory-streams-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092,broker3:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
props.put(StreamsConfig.producerPrefix(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), 60000);
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
props.put(StreamsConfig.STATE_DIR_CONFIG, "/var/lib/kafka-streams");

StreamsConfig.EXACTLY_ONCE_V2 is the constant for the string "exactly_once_v2".

Key points:

• REPLICATION_FACTOR_CONFIG=3 sets the replication factor for the internal changelog and repartition topics Streams creates. With EOS, durable changelogs are essential — a lost changelog partition means lost state.
• TRANSACTION_TIMEOUT_CONFIG (the producer’s transaction.timeout.ms) bounds how long a commit interval’s worth of work may take. The broker aborts any transaction that exceeds it. Raise it if a single commit interval can include long state-store flushes; keep it under transaction.max.timeout.ms.
• APPLICATION_ID_CONFIG must be unique per logical application. It seeds the consumer group.id, the internal topic names, and the transactional.id prefix. Reusing it across distinct applications collides on internal topics and transaction metadata.
• NUM_STANDBY_REPLICAS_CONFIG maintains warm standby copies of state stores, cutting recovery time after a rebalance at the cost of extra network and disk.

Implementing EOS with Spring Kafka

Spring Kafka provides EOS through a KafkaTransactionManager wired into the listener container. With a transaction manager present, the container starts a Kafka transaction before invoking the listener, sends the consumer offsets into that transaction, and commits on success or aborts on a thrown exception. There are two styles: container-managed transactions (the common case) and programmatic KafkaTemplate.executeInTransaction(). Both require a transaction-capable producer factory (one with a transactionIdPrefix).

With Spring Boot, setting spring.kafka.producer.transaction-id-prefix is enough — Boot auto-configures the KafkaProducerFactory with that prefix, a KafkaTransactionManager, and wires it into the listener container. The manual beans below show what Boot does for you.

Producer factory

@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 60000);
    DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(props);
    pf.setTransactionIdPrefix("spring-kafka-tx-");
    return pf;
}

Set the transactional ID prefix with setTransactionIdPrefix(...) rather than putting ProducerConfig.TRANSACTIONAL_ID_CONFIG directly in the map; the factory derives a per-instance transactional.id from this prefix. Under EOSMode.V2, Spring uses a single transactional.id per group/instance (it no longer needs a distinct ID per topic-partition as in V1), so the prefix must differ across application replicas — typically by including the instance ordinal.

Transaction manager and listener container

@Bean
public KafkaTransactionManager<String, String> transactionManager(
        ProducerFactory<String, String> producerFactory) {
    return new KafkaTransactionManager<>(producerFactory);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
        ConsumerFactory<String, String> consumerFactory,
        KafkaTransactionManager<String, String> transactionManager) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.getContainerProperties().setKafkaAwareTransactionManager(transactionManager);
    factory.getContainerProperties().setEosMode(ContainerProperties.EOSMode.V2);
    return factory;
}

• EOSMode.V2 is the only supported EOS mode in current Spring Kafka (it was once called BETA); it corresponds to Kafka’s exactly_once_v2 model. The legacy V1/ALPHA modes have been removed.
• Setting the transaction manager on the container properties is what makes the container transactional. The container begins the transaction, sends offsets into it, and commits or aborts around each listener invocation.
• Consumer offsets are committed atomically with the produced records via the producer’s sendOffsetsToTransaction.

Pitfall: @Transactional on a listener method does not by itself create the Kafka transaction. Without a transaction manager configured on the container, the container commits offsets normally and the annotation only governs whatever other (e.g. JDBC) transaction manager you reference — EOS is silently absent. Use the container KafkaTransactionManager to bound the Kafka transaction; use @Transactional only to chain an additional resource.

Consumer factory

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "spring-eos-consumer-group");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    return new DefaultKafkaConsumerFactory<>(props);
}

isolation.level=read_committed and enable.auto.commit=false are both required here. With container transactions Spring forces auto-commit off, but set it explicitly so the intent is clear. Without read_committed, downstream consumers would read aborted and uncommitted records, defeating EOS.

Operational Tuning and Resource Considerations

Transaction coordinator load

Every begin, commit, and abort writes to __transaction_state, and each commit fans out marker writes to all participating partitions. Under high transaction rates the coordinator and marker traffic become the limiting factor, so prefer fewer, larger transactions (longer commit intervals) over many tiny ones. If you suspect uneven load, confirm __transaction_state partitions are spread across brokers; the partition count (transaction.state.log.num.partitions, default 50) is fixed at topic creation and cannot be changed live.

Commit interval trade-off

In Kafka Streams, commit.interval.ms controls commit frequency. The default is 100 ms when processing.guarantee=exactly_once_v2 (it is 30000 ms only for at_least_once). A longer interval amortizes coordinator overhead but increases the volume of records reprocessed after a failure, since everything since the last commit is replayed. For Spring Kafka, the transaction scope is one listener invocation by default; a batch listener commits once per delivered batch.

State store and changelog writes

With EOS, Streams writes state changes and their changelog records within the transaction, so a store update and its durable changelog entry commit together. Monitor end-to-end commit latency via the Streams client metrics (below) rather than a non-existent per-store flush gauge; if commits are slow, check disk I/O on state.dir and the changelog topic’s replication health.

External system integration

Kafka EOS applies only to Kafka-to-Kafka flows. Writing to an external database or API is not covered by the Kafka transaction. Use one of:

• Transactional outbox: write the intended change to a Kafka topic inside the transaction, then have a sink connector (e.g. via CDC/Connect) apply it idempotently downstream.
• Idempotent writes: make the external write safe to repeat using a natural key, upsert, or a dedup table keyed by (topic, partition, offset).

Memory overhead

The transactional producer buffers records until commit and keeps per-partition state. If you hit buffer pressure (visible as elevated bufferpool-wait-time-ns-total or producer send blocking), raise buffer.memory and tune batch.size/linger.ms so batches flush efficiently.

Failure Scenarios and Recovery

Application crash

In-flight transactions are left open and are aborted by the coordinator once transaction.timeout.ms elapses (or immediately when the restarted instance re-initializes the transactional.id). On restart, initTransactions() bumps the epoch, fencing the dead instance with ProducerFencedException, and the application resumes from the last committed offset. Because offsets were committed atomically with output, only records since the last commit are reprocessed.

Broker failure (transaction coordinator)

The coordinator for a transactional.id lives on the leader of its __transaction_state partition. If that broker fails, leadership moves to an in-sync replica and no transaction metadata is lost. Producers transiently see NOT_COORDINATOR or COORDINATOR_NOT_AVAILABLE and automatically rediscover the new coordinator and retry.

Network partition

If a producer is isolated longer than transaction.timeout.ms, the coordinator aborts the open transaction. When connectivity returns, the now-fenced producer fails its in-flight calls; the framework re-initializes and starts a fresh transaction. Both Kafka Streams and Spring Kafka handle this recovery internally.

Zombie fencing during rebalance

When a Streams task moves to another instance, the new owner re-initializes the transactional.id and bumps the epoch, fencing any lingering transaction from the prior owner. This is the core mechanism that prevents duplicate output across rebalances.

Verifying transaction state

Use kafka-transactions.sh (added by KIP-664) to inspect a transactional producer’s state — useful for diagnosing a hung transaction that is blocking the LSO on a partition:

kafka-transactions.sh --bootstrap-server broker1:9092 \
  --describe --transactional-id spring-kafka-tx-0

This reports the producer ID, epoch, coordinator, current state, and transaction timeout. To find hanging transactions on a specific partition, use --find-hanging, and --abort to clear one. To enumerate all known transactional IDs, use --list.

Monitoring and Observability

Broker-side metrics

These MBean names are from the official Apache Kafka monitoring documentation. The transaction-coordinator gauges report over the last 30 seconds.

A high partition-load-time after a broker restart points at a large or slow-to-replicate __transaction_state; a rising VerificationFailureRate usually means zombie producers being fenced.

Producer / application metrics

Producer transaction metrics are cumulative nanosecond totals under producer-metrics, not rate gauges. Derive rates by differencing over time.

For Kafka Streams, the client-level MBean kafka.streams:type=stream-metrics,client-id=* exposes commit-total and commit-rate (alongside commit-latency-avg/-max), which are the right signals for commit health under EOS. Spring Kafka applications can publish client metrics to Micrometer by enabling micrometer on the listener container / producer factory (setMicrometerEnabled(true)).

Alerting thresholds

Debug logging

Enable DEBUG for org.apache.kafka.clients.producer.internals.TransactionManager during an initial rollout to trace transaction lifecycle events (begin, add-partitions, commit, abort, fencing). Revert to INFO afterward — this logger is high-volume under load.

Common Pitfalls

1. Assuming EOS is on by default. Streams’ processing.guarantee defaults to at_least_once. You must set exactly_once_v2 explicitly; otherwise everything below runs but the guarantee is at-least-once.

2. @Transactional without a container transaction manager. In Spring Kafka, the annotation alone does not start the Kafka transaction — without a KafkaTransactionManager on the container, EOS is silently absent. Verify boundaries with integration tests that inject failures (e.g. throw after a partial produce and assert no duplicate output).

3. Reusing a static transactional.id across instances. Two live producers sharing one ID continually fence each other. Let Spring derive it from a per-instance prefix and let Streams derive it from application.id; never pin a single static value across replicas.

4. Setting transaction.timeout.ms too low. If a commit interval’s work exceeds the timeout, the broker aborts mid-flight. Size it from measured per-commit duration under peak load (with headroom), and keep it under the broker’s transaction.max.timeout.ms.

5. Mixing transactional and non-transactional sends on one producer. Once a producer has a transactional.id, all of its sends must occur inside transactions. Use a separate, non-transactional producer for any fire-and-forget writes.

6. Skipping __transaction_state health checks. Confirm the internal topic’s replication and ISR:

  kafka-topics.sh --bootstrap-server broker1:9092 \
    --describe --topic __transaction_state

It should show replication factor ≥ 3 and min ISR satisfied; an under-replicated transaction log is a single point of EOS failure.

Conclusion

Exactly-once with Kafka Streams and Spring Kafka is production-proven, but it is a system-wide property, not a per-application toggle. Every stage must participate in the transaction or be designed for idempotent consumption. The operational essentials:

• Set processing.guarantee=exactly_once_v2 (Streams) or wire a KafkaTransactionManager with EOSMode.V2 (Spring Kafka) — it is never the default.
• Ensure isolation.level=read_committed on every consumer of transactional output (Streams sets this for its internal clients automatically).
• Keep __transaction_state at replication factor ≥ 3 with min.isr ≥ 2.
• Monitor the real metrics: transaction-coordinator-metrics partition-load time, AddPartitionsToTxnManager verification rate, Streams commit-latency/commit-rate, and consumer lag.
• Size transaction.timeout.ms from measured per-commit processing time under peak load, below transaction.max.timeout.ms.

For the original design, see KIP-98: Exactly Once Delivery and Transactional Messaging and KIP-447: Producer scalability for exactly once semantics (the basis of exactly_once_v2). The transaction tooling is described in KIP-664. See also the Spring for Apache Kafka transactions reference.