Unit Testing Kafka Streams Topologies with TopologyTestDriver

A Kafka Streams topology that silently drops records, mishandles state, or emits incorrect aggregates can corrupt downstream systems long before any alert fires. TopologyTestDriver, shipped in the kafka-streams-test-utils artifact, runs a compiled topology in-process with no broker, no controller/ZooKeeper, and no network. It is single-threaded and synchronous: you pipe records in, advance time explicitly, and assert on outputs and state-store contents immediately. This makes it the right tool for fast, deterministic unit tests of stream-processing logic.

This article covers the test lifecycle, input/output assertions, state-store inspection, and the one detail teams most often get wrong: how stream time versus wall-clock time drives windowing and suppression.

Why TopologyTestDriver Matters for Production Operations

In a production Stream Processing environment, confidence in your applications comes from repeatable verification of business logic. Integration tests against a real cluster are valuable but slow and flaky, and they couple every test run to broker availability and consumer rebalancing. TopologyTestDriver removes that coupling: it executes the same Topology object your application runs, but deterministically and in memory.

The testing approach is identical whether you build that topology with the high-level DSL or the lower-level Processor API, because the driver operates on the compiled Topology — a distinction explored in Kafka Streams DSL vs Processor API: When to Use Each.

Operational benefits:

  • No external dependencies: no broker, no controller, no Schema Registry.
  • Deterministic execution: no race conditions, no rebalances, no network timeouts.
  • Fast feedback: hundreds of topology tests run in seconds.
  • State-store inspection: query and assert on store contents mid-test.
  • Explicit time control: advance stream time and wall-clock time programmatically.

Setting Up the Test Dependencies

TopologyTestDriver lives in kafka-streams-test-utils. Pin its version to your kafka-streams version exactly; mixing versions causes runtime incompatibilities.

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams-test-utils</artifactId>
    <version>3.6.1</version>
    <scope>test</scope>
</dependency>

For Gradle:

testImplementation 'org.apache.kafka:kafka-streams-test-utils:3.6.1'

The examples here use JUnit 5 and AssertJ:

<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.10.1</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.assertj</groupId>
    <artifactId>assertj-core</artifactId>
    <version>3.24.2</version>
    <scope>test</scope>
</dependency>

The fluent TestInputTopic/TestOutputTopic API used throughout was introduced by KIP-470 in Kafka 2.4. It replaced the older ConsumerRecordFactory, pipeInput/readOutput, and OutputVerifier classes, which are deprecated. The current API is described in the Kafka Streams Testing guide.

Constructing a Test Driver and Supplying Input Records

The test lifecycle is predictable: build the topology, create the driver, create input/output topics, pipe records, assert, close. The driver is AutoCloseable, so a try-with-resources block guarantees cleanup.

flowchart TD A["Build topology"] --> B["Create TopologyTestDriver"] B --> C["Create input / output topics"] C --> D["pipeInput (records, timestamps)"] D --> E["Assert outputs and state stores"] E --> F["close (try-with-resources)"]

A minimal filter-and-transform test:

@Test
void shouldFilterAndTransformRecords() {
    StreamsBuilder builder = new StreamsBuilder();
    builder.<String, String>stream("input-topic")
           .filter((key, value) -> value != null && value.length() > 5)
           .mapValues(value -> value.toUpperCase())
           .to("output-topic");

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

    try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
        TestInputTopic<String, String> inputTopic =
            driver.createInputTopic("input-topic", new StringSerializer(), new StringSerializer());
        TestOutputTopic<String, String> outputTopic =
            driver.createOutputTopic("output-topic", new StringDeserializer(), new StringDeserializer());

        inputTopic.pipeInput("key1", "short");
        inputTopic.pipeInput("key2", "long enough");

        List<KeyValue<String, String>> results = outputTopic.readKeyValuesToList();
        assertThat(results).hasSize(1);
        assertThat(results.get(0).value).isEqualTo("LONG ENOUGH");
    }
}

Notes on the configuration:

  • BOOTSTRAP_SERVERS_CONFIG is required by StreamsConfig validation even though the driver never opens a connection — any dummy value works.
  • DEFAULT_KEY_SERDE_CLASS_CONFIG/DEFAULT_VALUE_SERDE_CLASS_CONFIG accept either a Class or a fully-qualified class-name string. Serdes.String().getClass() is the form used in the official docs; it resolves to org.apache.kafka.common.serialization.Serdes$StringSerde.
  • The serdes you pass to createInputTopic/createOutputTopic are independent of the topology’s default serdes — they only handle serialization at the test boundary.

You can pipe records with explicit timestamps via pipeInput(key, value, timestampMs); this is how you advance stream time, which matters for every windowed test below.

Asserting on State Stores

Beyond output topics, the driver exposes state stores directly via getKeyValueStore(name), getWindowStore(name), getSessionStore(name), and the generic getStateStore(name). This is essential for verifying stateful operators.

The DSL auto-generates store names (e.g. KSTREAM-AGGREGATE-STATE-STORE-0000000003) that are positional and change when you add or remove operators. Always name stores explicitly with Materialized.as(...) so tests reference a stable handle.

Here a tumbling-window count groups by the message key (the word) and inspects the window store. Note that all input records below fall in the same one-minute window [0, 60000):

@Test
void shouldAggregateWordCountsInWindow() {
    StreamsBuilder builder = new StreamsBuilder();
    builder.<String, String>stream("words")
           .groupByKey()
           .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
           .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("word-count-store"))
           .toStream()
           .to("counts");

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "word-count-test");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

    try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
        TestInputTopic<String, String> inputTopic =
            driver.createInputTopic("words", new StringSerializer(), new StringSerializer());

        // The KEY is the word being counted; the value is ignored by count().
        inputTopic.pipeInput("hello", "x", 1000L);
        inputTopic.pipeInput("hello", "x", 2000L);
        inputTopic.pipeInput("world", "x", 3000L);

        WindowStore<String, Long> store = driver.getWindowStore("word-count-store");
        // fetch(K key, long time) returns the value directly for the window containing `time`.
        long helloCount = store.fetch("hello", 1000L);
        long worldCount = store.fetch("world", 3000L);

        assertThat(helloCount).isEqualTo(2L);
        assertThat(worldCount).isEqualTo(1L);
    }
}

The single-timestamp overload is V fetch(K key, long time) — it returns the value directly (or null), not an iterator. The range overload WindowStoreIterator<V> fetch(K key, Instant from, Instant to) returns an iterator you must close, whose elements are KeyValue<Long, V> (window-start timestamp to value). Mixing these up is a common compile-time error.

Driving Windowed Operators: Stream Time vs Wall-Clock Time

This is the section worth slowing down for. Kafka Streams has two clocks, and they are not interchangeable:

  • Stream time is the maximum record timestamp the topology has observed. It advances only when you pipe records with timestamps (pipeInput(..., timestampMs)). Window closing, grace periods, and suppress(untilWindowCloses(...)) are all driven by stream time.
  • Wall-clock time is a separate mocked clock advanced with driver.advanceWallClockTime(Duration). It triggers only wall-clock punctuators (PunctuationType.WALL_CLOCK_TIME). It does not close windows or flush suppression buffers.

A frequent mistake is calling advanceWallClockTime to flush a suppress(untilWindowCloses(...)) buffer. That has no effect. To emit a final windowed result, you must push stream time past the window’s end (plus its grace period) by piping a later record — typically a “dummy” record with a timestamp far enough in the future.

The following test verifies that suppression emits exactly one final result per window. The 10-second window with no grace closes once stream time reaches 10000 ms, so a record at 20000 ms drives the emit:

@Test
void shouldSuppressUntilWindowCloses() {
    StreamsBuilder builder = new StreamsBuilder();
    builder.<String, String>stream("events")
           .groupByKey()
           .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(10)))
           .count()
           .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
           .toStream()
           .to("suppressed-output");

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "suppress-test");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

    try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
        TestInputTopic<String, String> inputTopic =
            driver.createInputTopic("events", new StringSerializer(), new StringSerializer());
        // Output key is Windowed<String>; deserialize the value (the count) as Long.
        TestOutputTopic<Windowed<String>, Long> outputTopic =
            driver.createOutputTopic(
                "suppressed-output",
                new TimeWindowedDeserializer<>(new StringDeserializer(), 10_000L),
                new LongDeserializer());

        // Two events for key "A" inside the window [0, 10000).
        inputTopic.pipeInput("A", "x", 1000L);
        inputTopic.pipeInput("A", "y", 5000L);

        // Nothing emitted yet: the window is still open.
        assertThat(outputTopic.isEmpty()).isTrue();

        // Advance STREAM time past the window end (10000) with a later record.
        inputTopic.pipeInput("A", "flush", 20_000L);

        List<KeyValue<Windowed<String>, Long>> results = outputTopic.readKeyValuesToList();
        assertThat(results).hasSize(1);
        assertThat(results.get(0).value).isEqualTo(2L);
    }
}

Because the output of a windowed aggregation is keyed by Windowed<K>, the output topic must deserialize keys with TimeWindowedDeserializer, constructed with the inner key deserializer and the window size. Use advanceWallClockTime only when you are testing a wall-clock punctuator you registered yourself, as shown next.

Testing Processor API Topologies

With the Processor API you assemble a Topology from sources, processors, and sinks, attach state stores with addStateStore(...), and forward records explicitly with context.forward(...). The driver usage is unchanged.

The processor below maintains a running sum per key in an in-memory store and forwards the new total on each record:

@Test
void shouldProcessWithCustomProcessor() {
    Topology topology = new Topology();
    topology.addSource("Source", "input-topic")
            .addProcessor("MyProcessor", MyRunningSumProcessor::new, "Source")
            .addStateStore(
                Stores.keyValueStoreBuilder(
                    Stores.inMemoryKeyValueStore("my-store"),
                    Serdes.String(),
                    Serdes.Integer()
                ),
                "MyProcessor"
            )
            .addSink("Sink", "output-topic", "MyProcessor");

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "processor-test");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

    try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
        TestInputTopic<String, Integer> inputTopic =
            driver.createInputTopic("input-topic", new StringSerializer(), new IntegerSerializer());
        TestOutputTopic<String, Integer> outputTopic =
            driver.createOutputTopic("output-topic", new StringDeserializer(), new IntegerDeserializer());

        inputTopic.pipeInput("A", 10);
        inputTopic.pipeInput("A", 20);

        KeyValueStore<String, Integer> store = driver.getKeyValueStore("my-store");
        assertThat(store.get("A")).isEqualTo(30);

        List<KeyValue<String, Integer>> results = outputTopic.readKeyValuesToList();
        assertThat(results).extracting(KeyValue::value).containsExactly(10, 30);
    }
}

Use serializers at the test boundary that match the types the processor expects: here the source carries Integer values, so the input topic uses IntegerSerializer. Asserting on the store and the output topic catches two distinct bugs — incorrect state and missing forward calls. The DSL vs Processor API guide covers when to reach for this lower-level API.

Integrating Topology Tests into CI/CD Pipelines

Because the driver has no external dependencies, these tests run in any JVM environment — CI runners, containers, or Kubernetes pods — with a standard build invocation:

mvn test -pl kafka-streams-app -Dtest=TopologyUnitTests

To gate merges on coverage, bind JaCoCo. The check goal only sees data if prepare-agent runs first (it wires the agent into Surefire); without it, check evaluates empty coverage and silently passes. Bind both, and run mvn verify:

<plugin>
    <groupId>org.jacoco</groupId>
    <artifactId>jacoco-maven-plugin</artifactId>
    <version>0.8.11</version>
    <executions>
        <execution>
            <id>prepare-agent</id>
            <goals>
                <goal>prepare-agent</goal>
            </goals>
        </execution>
        <execution>
            <id>check</id>
            <phase>verify</phase>
            <goals>
                <goal>check</goal>
            </goals>
            <configuration>
                <rules>
                    <rule>
                        <element>PACKAGE</element>
                        <limits>
                            <limit>
                                <counter>LINE</counter>
                                <value>COVEREDRATIO</value>
                                <minimum>0.85</minimum>
                            </limit>
                        </limits>
                    </rule>
                </rules>
            </configuration>
        </execution>
    </executions>
</plugin>

TopologyTestDriver does not touch Schema Registry. If your application uses Avro or Protobuf, configure the serdes against a mock registry URL (mock://) so tests stay self-contained; pair these unit tests with separate integration tests that validate real serialization compatibility. See the Confluent Schema Registry documentation.

Common Pitfalls

Wall-clock time does not close windows. As shown above, advanceWallClockTime triggers only wall-clock punctuators. Window finalization, grace periods, and suppression are driven by stream time, which you advance by piping records with later timestamps.

Auto-generated store names are fragile. DSL store names are positional; adding or removing an operator can rename them and break getWindowStore/getKeyValueStore lookups. Name stores explicitly with Materialized.as(...) or Stores.inMemoryKeyValueStore("...").

Window-keyed output needs TimeWindowedDeserializer. A windowed aggregation produces keys of type Windowed<K>; the output topic’s key deserializer must be a TimeWindowedDeserializer built with the inner deserializer and window size, or reads will fail.

Serde mismatches surface as ClassCastException. The serdes passed to createInputTopic/createOutputTopic must match the actual record types flowing across that boundary, independent of the topology’s default serdes.

Always close the driver. Use try-with-resources or a finally block. A leaked driver retains in-memory state (and RocksDB resources if configured) that can corrupt later tests.

Forgetting context.forward(). In Processor API code, a processor that never forwards produces no output. Asserting on the output topic — not just the store — catches this immediately.

Embedding these tests in your workflow shifts validation left, catching logic errors before staging or production and keeping your Stream Processing footprint stable as it grows across teams.

Back to Kafka Streams DSL vs Processor API When To Use Each