Achieving Exactly-Once Semantics in Kafka Streams with Spring Cloud Stream

For platform and SRE engineers running Apache Kafka in production, the distinction between at-least-once and exactly-once is not academic—it directly impacts data integrity, reconciliation cost, and on-call burden. When you build stream processors with Spring Cloud Stream and the Kafka Streams binder, exactly-once semantics (EOS) is a single configuration flip, but operating it correctly requires understanding what the guarantee actually covers and how to verify it.

This guide is an operational reference for configuring, validating, and troubleshooting EOS in Spring Cloud Stream applications backed by Kafka Streams. It covers transactional offset management, idempotent writes, the Spring binder properties that turn EOS on, state-store recovery, the real JMX metrics you can monitor, and the failure modes you will actually hit.

What Kafka’s Exactly-Once Guarantee Covers

Kafka’s EOS is scoped to the read-process-write cycle: consuming records from input topics, transforming them, updating state stores, and producing to output topics. Kafka Streams binds all of these into a single transaction. If a failure occurs mid-processing, the transaction aborts, the input offsets are not committed, and the records are reprocessed—without the aborted output ever becoming visible to downstream consumers.

This atomicity rests on two primitives:

  • Idempotent producers (enable.idempotence=true) deduplicate producer retries within a single partition using a producer ID and per-partition sequence numbers. Kafka Streams sets this automatically when EOS is enabled.
  • Transactions extend the guarantee across multiple partitions and across the offset-commit-plus-output-write boundary. Kafka Streams sets the consumer’s isolation.level=read_committed automatically, so a consumer never sees records from an aborted transaction.

The boundary matters operationally: EOS does not extend to external side effects such as a database write or a REST call. For those, use the transactional outbox pattern. For the state-store mechanics that underpin transactional recovery, see Stateful Stream Processing with Kafka Streams, and Stream Processing for the broader architectural context.

Enabling Exactly-Once in Spring Cloud Stream

Kafka Streams exposes the processing guarantee through the processing.guarantee config, which accepts at_least_once (default) or exactly_once_v2. The value exactly_once_v2 selects the scalable EOS implementation from KIP-447: the broker-side changes ship in Kafka 2.5+, and the config value itself was renamed to exactly_once_v2 in Kafka 2.8 (previously exactly_once_beta). The original exactly_once (eos-alpha) and exactly_once_beta values were deprecated in 3.0 and removed in 4.0 per KIP-732. On any current cluster, use exactly_once_v2.

With the Kafka Streams binder, pass Kafka Streams configs through spring.cloud.stream.kafka.streams.binder.configuration.*. The binder automatically applies the EOS-implied consumer (read_committed) and producer (enable.idempotence=true) settings.

spring:
  cloud:
    function:
      definition: process
    stream:
      kafka:
        streams:
          binder:
            brokers: kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092
            configuration:
              processing.guarantee: exactly_once_v2
              # Replication factor for internal (changelog/repartition) topics.
              # Use at least 3 in production so EOS survives a broker loss.
              replication.factor: 3
              num.standby.replicas: 1
      bindings:
        process-in-0:
          destination: input-topic
          consumer:
            use-native-decoding: true
        process-out-0:
          destination: output-topic
          producer:
            use-native-encoding: true
            partition-count: 6

The replication.factor here applies to the internal topics Kafka Streams creates (changelog and repartition topics); set it to at least 3 so the changelog—and therefore your recoverable state—survives a broker failure. num.standby.replicas: 1 keeps a warm copy of each state store so failover does not require a full restore.

Transactional Mechanics and Offset Management

When EOS is enabled, each Kafka Streams task wraps its work in a transaction: read a batch from the input topic, run the topology, update local state stores, and produce output. The transaction commits atomically, writing the output records and the consumed offsets together. Offsets are not committed by the consumer directly; Kafka Streams writes them as part of the transaction via sendOffsetsToTransaction, so offset advancement and output visibility are inseparable.

Under exactly_once_v2, a single producer per StreamThread (rather than one per input partition) handles all of that thread’s tasks—this is the scalability improvement over the original eos-alpha protocol. Each producer uses a transactional.id derived from the application.id and an assigned identity. On restart after a crash, the new producer claims the same transactional identity and fences the old one, so any zombie cannot commit a partial transaction.

Because consumers read read_committed, reprocessing after an abort never exposes the aborted output. To inspect the consumer group:

kafka-consumer-groups --bootstrap-server kafka-broker-1:9092 \
  --group my-spring-app \
  --describe

This shows current offsets, log-end offsets, and lag. It does not expose transactional state—there is no --verbose flag for --describe. To inspect open or hanging transactions, use the kafka-transactions tool (KIP-664):

# List active transactions known to the coordinators
kafka-transactions --bootstrap-server kafka-broker-1:9092 list

# Detect transactions hanging on a specific topic
kafka-transactions --bootstrap-server kafka-broker-1:9092 \
  find-hanging --topic output-topic

State Store Recovery and Failure Scenarios

State stores are the component most exposed to failure. When a task fails, its state store must be restored to a point consistent with the committed offsets. Kafka Streams does this through a per-store changelog topic; with EOS, changelog writes are part of the same transaction as the output records, so the restored state is guaranteed to align with what downstream consumers have seen.

During recovery the task replays its changelog up to the last committed offset and rebuilds the store. For large stores this can take minutes, which is why num.standby.replicas matters: a standby continuously replays the changelog, so on failover the active task resumes from a near-current copy instead of restoring from scratch.

To watch restore progress, read the Kafka Streams state-store JMX metrics. These are at recording level DEBUG, so set metrics.recording.level: DEBUG to populate them:

kafka-run-class kafka.tools.JmxTool \
  --object-name "kafka.streams:type=stream-state-metrics,thread-id=*,task-id=*,*" \
  --attributes restore-rate,restore-latency-avg,restore-latency-max \
  --jmx-url service:jmx:rmi:///jndi/rmi://localhost:9999/jmxrmi

restore-rate is the average number of restore operations per second; restore-latency-avg/restore-latency-max give restore execution time in nanoseconds. (Note: there is no restoration-total/restoration-rate attribute—those names do not exist.) If restore is slow, raise num.stream.threads or add application instances to spread changelog partitions across more consumers.

Performance Implications and Tuning

EOS adds overhead: each commit is a transaction requiring round trips to the transaction coordinator, and the producer must flush before committing. Expect measurably higher end-to-end latency than at-least-once, dominated by how often you commit.

The single most important EOS-specific default to understand is the commit interval. When processing.guarantee is exactly_once_v2 (or the deprecated exactly_once), the default commit.interval.ms is 100 ms, not the 30,000 ms used for at-least-once. Kafka Streams chooses a tight interval under EOS because each commit makes output visible to read_committed consumers—so the commit interval is effectively your output latency floor, and it bounds how much work is replayed after an abort.

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            configuration:
              processing.guarantee: exactly_once_v2
              # Default under EOS is 100ms; raise it to trade latency for throughput.
              commit.interval.ms: 1000
              # Must exceed the worst-case time to process and commit one batch.
              # Default is 60000 (60s).
              transaction.timeout.ms: 60000
              # Record cache; larger cache = fewer changelog/state writes.
              statestore.cache.max.bytes: 10485760
              max.task.idle.ms: 10000

Notes on these keys:

  • transaction.timeout.ms (default 60000) must be larger than the maximum time to process and commit a batch, and it must not exceed the broker’s transaction.max.timeout.ms (broker default 900000). If a task stalls longer than this, the coordinator aborts the transaction.
  • statestore.cache.max.bytes controls the record cache that deduplicates and batches writes before they hit RocksDB and the changelog. (cache.max.bytes.buffering is the deprecated alias for the same setting—prefer the newer key.) A larger cache means fewer changelog writes per commit at the cost of memory.
  • Lengthening commit.interval.ms increases throughput but raises output latency and the amount of reprocessing after a failure.

To find transactions that are running long on the broker side, monitor:

kafka-run-class kafka.tools.JmxTool \
  --object-name "kafka.server:type=ReplicaManager,name=PartitionsWithLateTransactionsCount" \
  --jmx-url service:jmx:rmi:///jndi/rmi://kafka-broker-1:9999/jmxrmi

PartitionsWithLateTransactionsCount reports partitions with open transactions exceeding transaction.max.timeout.ms—a sustained non-zero value points to stuck producers or timeouts that need investigation. (There is no kafka.server:type=TransactionCoordinatorMetrics MBean with ActiveTransactionCount/TransactionAbortRate; those metric names are not part of Kafka.)

Validating Exactly-Once Behavior

Validate EOS before production by inducing failure and confirming no duplicates appear downstream—ideally as controlled chaos testing in staging. Produce a known sequence of uniquely identified records, kill a broker or the application mid-stream, and after recovery confirm each identifier appears exactly once on the output topic.

# Produce 1000 test records with unique IDs
for i in $(seq 1 1000); do
  echo "{\"id\": $i, \"value\": \"test-$i\"}" | \
    kafka-console-producer --bootstrap-server kafka-broker-1:9092 \
    --topic input-topic
done

# After inducing failure and recovery, count distinct IDs in the output.
# --isolation-level read_committed ensures aborted output is excluded.
kafka-console-consumer --bootstrap-server kafka-broker-1:9092 \
  --topic output-topic \
  --from-beginning \
  --isolation-level read_committed \
  --timeout-ms 30000 | \
  jq -r '.id' | sort -n | uniq | wc -l

The distinct count should equal the number of input records. If it does not, inspect the consumer group offsets and broker transaction state. The Apache Kafka documentation on exactly-once delivery and the Kafka Streams developer guide cover the guarantees and their limits.

Troubleshooting Common EOS Issues

Transaction timeouts. A transaction that exceeds transaction.timeout.ms is aborted by the coordinator; the original producer is then fenced and surfaces a ProducerFencedException or InvalidProducerEpochException in the logs. Increase transaction.timeout.ms (while staying under the broker’s transaction.max.timeout.ms) or reduce per-batch processing time.

State-store corruption from under-replication. If a changelog topic’s replication factor is too low, a broker loss can lose changelog data, leaving a store that cannot be fully restored and a task stuck in an error state. Keep replication.factor at 3+ for internal topics and alarm on the broker UnderReplicatedPartitions metric.

Zombie fencing during rolling restarts. When a task restarts, it reclaims its transactional identity and fences the prior producer. A ProducerFencedException here is expected and desirable—the old producer cleanly shuts down. The hazard is alerting noise: make sure your alerts distinguish routine fencing during deploys from fencing caused by, for example, two instances sharing an application.id.

For how state stores and transactions interact in detail, see Stateful Stream Processing with Kafka Streams. For the protocol itself, KIP-447 specifies the scalable exactly_once_v2 producer model.

With the right processing.guarantee, sane internal-topic replication, a commit interval tuned to your latency budget, and monitoring built on metrics that actually exist, Spring Cloud Stream plus Kafka Streams delivers genuine exactly-once processing in production.

Back to Stateful Stream Processing With Kafka Streams