Enabling Exactly‑Once Semantics in Kafka Streams: Configuration, Code, and Pitfalls

Exactly‑Once Semantics in Kafka Streams

When a stream processor receives duplicate records, or the application crashes mid‑processing, the default at‑least‑once guarantee can leave the state and output topic in an inconsistent state. Exactly‑once semantics (EOS) solve this by ensuring that each input record is processed, written to the state store, and emitted to downstream topics exactly one time, even across failures.

Enabling EOS with StreamsConfig

EOS is a feature of the Kafka Streams library, not the broker itself. To activate it you only need to set the processing.guarantee property. For Kafka Streams 3.x the recommended value is exactly_once_v2 (the older exactly_once is still accepted for backward compatibility).

Property	Value	Explanation
StreamsConfig.APPLICATION_ID_CONFIG	eos-demo	Unique ID for the Streams instance.
StreamsConfig.BOOTSTRAP_SERVERS_CONFIG	localhost:9092	Kafka cluster address.
StreamsConfig.PROCESSING_GUARANTEE_CONFIG	exactly_once_v2	Enables transactional processing and idempotent producer.
StreamsConfig.TRANSACTIONAL_ID_PREFIX	eos-demo-	Optional prefix for transaction IDs; required if you want to control the prefix.

Minimal Java Example

Below is a compact Streams application that counts the occurrences of keys in an input topic and writes the result to an output topic. The state store is changelogged, so the transaction includes both the state update and the output record.

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.util.Properties;

public class EosDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "eos-demo");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
        // Optional: set a prefix to keep transaction IDs predictable
        props.put(StreamsConfig.TRANSACTIONAL_ID_PREFIX_CONFIG, "eos-demo-");

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> input = builder.stream("input-topic");

        KTable<String, Long> counts = input
            .groupByKey()
            .count(Materialized.as("counts-store"));

        counts.toStream()
            .to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        // Add shutdown hook to cleanly close the Streams instance
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

Running this program will:

• Start a transactional producer that is automatically idempotent.
• Wrap each batch of state updates and output records in a single transaction.
• Write the counts-store changelog to a Kafka topic named counts-store-changelog (created automatically).

Limits and Common Pitfalls

• Broker version: EOS requires broker 2.5 or newer to support the transaction log.
• State store type: Only changelogged state stores participate in EOS. In‑memory stores, or stores that are explicitly disabled for changelogging, will cause the transaction to fail.
• Transactional producer: The Streams library automatically creates a transactional producer when EOS is enabled, but you must not override it or set conflicting properties that disable idempotence.
• Multiple state stores: If your topology uses several state stores, each must have a changelog topic. Missing changelog topics will abort the transaction.
• Compatibility with older clients: Mixing an EOS Streams app with older Kafka clients that do not support the newer transaction APIs can lead to broker errors.
• Performance overhead: Transactions require extra writes to the __transaction_state log and additional network round‑trips. Measure latency and throughput before enabling EOS in a production environment.
• Transaction ID prefix: If you omit transaction.id.prefix, Kafka Streams generates a random prefix. In a multi‑instance deployment, this can lead to duplicate transaction IDs and failures.

How to Verify EOS Works

Below are two practical ways to confirm that your Streams application is truly processing each record once.

1. Manual Duplicate Test

1. Start the Streams application.
2. Produce a batch of records to input-topic using the console producer:

   kafka-console-producer --broker-list localhost:9092 --topic input-topic
   > key1 value1
   > key2 value2
   > key1 value1
   

3. Stop the application, restart it, and verify that the output-topic contains the correct counts (e.g., key1 appears only once).
4. Check the counts-store-changelog topic to ensure that duplicate records did not create duplicate entries.

2. Integration Test with Test Driver

Use the TopologyTestDriver to inject duplicate records and assert the final state.

Topology topology = builder.build();
try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
    TestInputTopic<String, String> input = driver.createInputTopic(
        "input-topic", Serdes.String().serializer(), Serdes.String().serializer());
    TestOutputTopic<String, Long> output = driver.createOutputTopic(
        "output-topic", Serdes.String().deserializer(), Serdes.Long().deserializer());

    input.pipeInput("key1", "val1");
    input.pipeInput("key1", "val1"); // duplicate

    List<KeyValue<String, Long>> results = output.readKeyValuesToList();
    assertEquals(1, results.size());
    assertEquals(1L, results.get(0).value); // only one increment
}

Because the test driver runs in a single JVM, it does not exercise the full transaction log, but it still verifies that the application logic does not double‑count.

Practical Checklist Before Enabling EOS

• Confirm broker and Streams library versions: kafka-broker --version and mvn dependency:tree | grep kafka-streams.
• Ensure all state stores are changelogged: check the topology with topology.describe().
• Set a transaction.id.prefix if you run multiple instances to avoid ID collisions.
• Run a small throughput test: produce 10,000 records, restart the app, and verify counts.
• Monitor the __transaction_state topic size; it can grow if the application is long‑running and not compacted.

When you follow these steps, you can confidently deploy a Kafka Streams application that guarantees exactly‑once processing for stateful operators, reducing the risk of duplicate data in downstream systems.