Testing topologies

Unit-test a StoatFlow topology in-memory with TopologyTestDriver — pipe records, advance time and watermarks, read state stores, and drive YAML config through StoatFlowTestDriver. No broker required.

stoatflow-test-utils runs your topology synchronously, in-process, with no Kafka broker. You pipe records in, read records out, advance time and watermarks by hand, and read state stores directly for assertions. Because processing is synchronous and single-threaded, tests are deterministic and easy to debug.

Add the dependency

stoatflow-test-utils is a testImplementation dependency. If you load configuration from application.yaml in tests (see Config-driven tests), also add stoatflow-test-runtime.

// build.gradle.kts
dependencies {
    testImplementation("io.stoatflow:stoatflow-test-utils:<stoatflow-version>")
    // Config-driven tests (loads application.yaml):
    testImplementation("io.stoatflow:stoatflow-test-runtime:<stoatflow-version>")
}

<!-- pom.xml -->
<dependency>
  <groupId>io.stoatflow</groupId>
  <artifactId>stoatflow-test-utils</artifactId>
  <version>${stoatflow.version}</version>
  <scope>test</scope>
</dependency>
<!-- Config-driven tests (loads application.yaml): -->
<dependency>
  <groupId>io.stoatflow</groupId>
  <artifactId>stoatflow-test-runtime</artifactId>
  <version>${stoatflow.version}</version>
  <scope>test</scope>
</dependency>

The current version is 1.0.0-alpha.16. See Installation for the repository and version setup.

A first test

TopologyTestDriver.fromBuilder(builder) compiles the topology from a StreamsBuilder and wires up the processor and state-store registries automatically — it's the recommended entry point. Then createInputTopic / createOutputTopic give you typed handles for piping records and reading results.

The serdes you pass to createInputTopic / createOutputTopic are applied, exactly as the engine would: piped keys and values are serialized to wire bytes and re-deserialized through the source-node serde before the first processor sees them, and sink output is deserialized through the serdes you pass at read time. (KS-shaped createInputTopic(topic, Serializer, Serializer) / createOutputTopic(topic, Deserializer, Deserializer) overloads exist too.) Records flow through the topology synchronously on pipeInput — by the time pipeInput returns, the output is already on the sink queue.

import io.stoatflow.core.topology.StreamsBuilder
import io.stoatflow.testutils.TopologyTestDriver
import org.apache.kafka.common.serialization.Serdes
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.api.BeforeEach
import org.junit.jupiter.api.Test

class UppercaseTopologyTest {
    private lateinit var driver: TopologyTestDriver

    @BeforeEach
    fun setup() {
        val builder = StreamsBuilder()
        builder
            .stream<String, String>("input-topic")
            .mapValues { value -> value.uppercase() }
            .filter { _, value -> value.length > 3 }
            .to("output-topic")

        driver = TopologyTestDriver.fromBuilder(builder, TopologyTestDriver.STRING_CONFIG)
    }

    @AfterEach
    fun teardown() = driver.close()

    @Test
    fun `maps and filters values`() {
        val input = driver.createInputTopic("input-topic", Serdes.String(), Serdes.String())
        val output = driver.createOutputTopic<String, String>("output-topic", Serdes.String(), Serdes.String())

        input.pipeInput("k1", "hello")
        input.pipeInput("k2", "no")   // filtered out (length <= 3 after uppercasing)

        assertThat(output.readRecord()?.value).isEqualTo("HELLO")
        assertThat(output.isEmpty()).isTrue()
    }
}

import io.stoatflow.core.topology.StreamsBuilder;
import io.stoatflow.testutils.TestInputTopic;
import io.stoatflow.testutils.TestOutputTopic;
import io.stoatflow.testutils.TopologyTestDriver;
import org.apache.kafka.common.serialization.Serdes;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import static org.assertj.core.api.Assertions.assertThat;

class UppercaseTopologyTest {

    private TopologyTestDriver driver;

    @BeforeEach
    void setup() {
        StreamsBuilder builder = new StreamsBuilder();
        builder.<String, String>stream("input-topic")
            .mapValues(value -> value.toUpperCase())
            .filter((key, value) -> value.length() > 3)
            .to("output-topic");

        driver = TopologyTestDriver.fromBuilder(builder, TopologyTestDriver.STRING_CONFIG);
    }

    @AfterEach
    void teardown() {
        driver.close();
    }

    @Test
    void mapsAndFiltersValues() {
        TestInputTopic<String, String> input =
            driver.createInputTopic("input-topic", Serdes.String(), Serdes.String());
        TestOutputTopic<String, String> output =
            driver.createOutputTopic("output-topic", Serdes.String(), Serdes.String());

        input.pipeInput("k1", "hello");
        input.pipeInput("k2", "no");   // filtered out

        assertThat(output.readRecord().getValue()).isEqualTo("HELLO");
        assertThat(output.isEmpty()).isTrue();
    }
}

TopologyTestDriver is not thread-safe — it is designed for single-threaded test execution. Create a fresh instance per test case (@BeforeEach) and close() it in @AfterEach when running parallel test runners.

Picking a config

fromBuilder(builder) defaults to ByteArray key/value serdes (matching the production StreamsConfig defaults). For String-keyed topologies, pass the bundled TopologyTestDriver.STRING_CONFIG, which sets String default serdes. For anything else, build a StreamsConfig explicitly, or use StoatFlowTestDriver.fromConfig(...) to load your real application.yaml (see below).

fromBuilder is overloaded with an optional initialWallClockTime (an Instant) — useful when a test asserts on absolute wall-clock-based fire times of scheduled sources or processing-time timers.

Piping input

TestInputTopic pipes records into a source topic. The simplest form takes a key and value; overloads add an explicit timestamp and headers, and a TestRecord form.

import io.stoatflow.core.state.KeyValue

val input = driver.createInputTopic("input-topic", Serdes.String(), Serdes.String())

input.pipeInput("k1", "v1")                       // uses the topic's current timestamp
input.pipeInput("k1", "v2", timestamp = 1_000L)   // explicit event time (ms)
input.pipeInput("k1", null)                        // tombstone (null value)

// Bulk helpers (KeyValue, matching Kafka Streams)
input.pipeKeyValueList(listOf(KeyValue("k1", "a"), KeyValue("k2", "b")))
input.pipeValueList(listOf("x", "y"))              // null keys

// Auto-advancing timestamps: each record steps 1s forward
input.withTimestamp(0L).withAutoAdvance(1_000L)
input.pipeInput("k1", "t0")   // ts = 0
input.pipeInput("k1", "t1")   // ts = 1000

import io.stoatflow.core.state.KeyValue;

TestInputTopic<String, String> input =
    driver.createInputTopic("input-topic", Serdes.String(), Serdes.String());

input.pipeInput("k1", "v1");                  // uses the topic's current timestamp
input.pipeInput("k1", "v2", 1_000L);          // explicit event time (ms)
input.pipeInput("k1", null);                  // tombstone (null value)

// Bulk helpers (KeyValue, matching Kafka Streams)
input.pipeKeyValueList(List.of(KeyValue.pair("k1", "a"), KeyValue.pair("k2", "b")));
input.pipeValueList(List.of("x", "y"));       // null keys

// Auto-advancing timestamps: each record steps 1s forward
input.withTimestamp(0L).withAutoAdvance(1_000L);
input.pipeInput("k1", "t0");   // ts = 0
input.pipeInput("k1", "t1");   // ts = 1000

withTimestamp and withAutoAdvance mutate state on the TestInputTopic instance that persists across calls. Call reset() to restore defaults (auto-advance disabled, timestamp back to the driver's current wall-clock time) if you reuse one input topic across test cases.

The event timestamp you pass to pipeInput is the record's event time — it drives windowing, joins, and watermark-based logic. If the source was configured with a WatermarkStrategy that carries a timestamp assigner, the driver applies it the same way the runtime does, overriding the passed timestamp. See Event time and watermarks.

Reading output

TestOutputTopic drains records the topology produced to a sink topic. readRecord() returns a TestRecord (or null when empty); there are convenience readers for keys, values, and bulk drains.

val output = driver.createOutputTopic<String, Long>("output-topic", Serdes.String(), Serdes.Long())

val record = output.readRecord()        // TestRecord<String, Long>? — null if empty
record?.key                              // "k1"
record?.value                            // 3L
record?.timestamp                        // event time of the emission

output.readValue()                       // next value only
output.readKeyValue()                    // next KeyValue (or null if empty)
output.readRecordsToList()               // drain all remaining as a list of TestRecord
output.readKeyValuesToList()             // drain all remaining as a list of KeyValues
output.readKeyValuesToMap()              // drain all; last value per key wins
output.getQueueSize()                    // records currently waiting
output.isEmpty()                         // true when drained

TestOutputTopic<String, Long> output =
    driver.createOutputTopic("output-topic", Serdes.String(), Serdes.Long());

TestRecord<String, Long> record = output.readRecord();   // null if empty
record.getKey();        // "k1"
record.getValue();      // 3L
record.getTimestamp();  // event time of the emission

output.readValue();              // next value only
output.readKeyValue();           // next KeyValue (or null if empty)
output.readRecordsToList();      // drain all remaining as a list of TestRecord
output.readKeyValuesToList();    // drain all remaining as a list of KeyValues
output.readKeyValuesToMap();     // drain all; last value per key wins
output.getQueueSize();           // records currently waiting
output.isEmpty();                // true when drained

Type parameters on TestOutputTopic are erased at runtime — they are not validated. If the topology emits a different type than declared, you get a ClassCastException only when you read a property. Match the declared K/V to the actual sink output types.

A common assertion for a KTable-backed aggregation is the changelog stream: count() / aggregate() emit the running result per key, so readKeyValuesToMap() (last-value-per-key) is the idiomatic way to assert the final state.

Advancing time

Stateful time-based logic — windowed aggregations with OnWindowClose, suppress, punctuators, timers, scheduled sources — fires when the relevant clock advances. By default the event-time clock advances automatically to each piped record's timestamp (matching Kafka Streams' TopologyTestDriver), so piping a record past a window's close already flushes it. You can also drive either clock manually:

advanceWatermark(timestampMs) advances event time. This fires event-time timers and STREAM_TIME punctuators, closes windows past their end + grace, flushes suppressed results, and fires STREAM_TIME scheduled sources. The watermark is monotonic — by default a backward request is ignored (a silent no-op), since each pipeInput already drives event time forward; an explicit call is only needed to push event time beyond the last piped record.
advanceWallClockTime(Duration) advances processing time. This fires processing-time timers and WALL_CLOCK_TIME punctuators, fires WALL_CLOCK_TIME (and cron) scheduled sources, and triggers a barrier commit (which flushes any caching-store emissions — see commitBarrier()).

Set test.driver.commit-per-pipe=false in your config Properties to opt out of the per-pipe parity: pipes then neither commit the cache nor advance event time, and you control both explicitly via commitBarrier() / advanceWatermark() (where a backward request throws, the legacy manual-control mode).

A windowed aggregation that only emits on window close needs event time pushed past the window's end plus its grace period — either by piping a later-timestamped record, or with an explicit advanceWatermark:

// Window [0, 5min), grace 30s — emits only on close
val builder = StreamsBuilder()
builder
    .stream<String, Int>("input")
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30)))
    .emitStrategy(EmitStrategy.onWindowClose())
    .count()
    .toStream()
    .map { windowedKey, count -> KeyValue(windowedKey.key, count) }
    .to("output")

val driver = TopologyTestDriver.fromBuilder(builder, TopologyTestDriver.STRING_CONFIG)
val input = driver.createInputTopic("input", Serdes.String(), Serdes.Integer())
val output = driver.createOutputTopic<String, Long>("output", Serdes.String(), Serdes.Long())

input.pipeInput("k1", 1, timestamp = 1_000L)
input.pipeInput("k1", 2, timestamp = 2_000L)

assertThat(output.isEmpty()).isTrue()           // window not closed yet

// Advance past window end (5min) + grace (30s) = 330s
driver.advanceWatermark(330_001L)

assertThat(output.readRecord()?.value).isEqualTo(2L)   // two records counted

// Window [0, 5min), grace 30s — emits only on close
StreamsBuilder builder = new StreamsBuilder();
builder.<String, Integer>stream("input")
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30)))
    .emitStrategy(EmitStrategy.onWindowClose())
    .count()
    .toStream()
    .map((windowedKey, count) -> new KeyValue<>(windowedKey.getKey(), count))
    .to("output");

TopologyTestDriver driver = TopologyTestDriver.fromBuilder(builder, TopologyTestDriver.STRING_CONFIG);
TestInputTopic<String, Integer> input =
    driver.createInputTopic("input", Serdes.String(), Serdes.Integer());
TestOutputTopic<String, Long> output =
    driver.createOutputTopic("output", Serdes.String(), Serdes.Long());

input.pipeInput("k1", 1, 1_000L);
input.pipeInput("k1", 2, 2_000L);

assertThat(output.isEmpty()).isTrue();           // window not closed yet

// Advance past window end (5min) + grace (30s) = 330s
driver.advanceWatermark(330_001L);

assertThat(output.readRecord().getValue()).isEqualTo(2L);   // two records counted

For session windows and suppress, the runtime pattern is the same one the example apps use: push the watermark past the inactivity gap (plus grace) to close the session and let suppress emit its final result. getCurrentWatermark() and getWallClockTime() return the current clocks for assertions.

Reading state stores

For stateful topologies you can read the state store directly instead of (or in addition to) asserting on output records. The accessor matches the store type the DSL operator created.

Accessor	Returns	For
`getKeyValueStore(name)`	`KeyValueStore<K, V>`	`count`, `reduce`, `aggregate`, materialized `KTable`
`getWindowStore(name)`	`WindowStore<K, V>`	time / sliding windowed aggregations
`getSessionStore(name)`	`SessionStore<K, V>`	session windowed aggregations
`getTimestampedKeyValueStore(name)`	`TimestampedKeyValueStore<K, V>`	when you also want the value's event timestamp
`getTimestampedWindowStore(name)`	`TimestampedWindowStore<K, V>`	windowed, with value timestamps
`getVersionedKeyValueStore(name)`	`VersionedKeyValueStore<K, V>`	versioned (point-in-time) stores
`getStateStore(name)`	`S : StateStore`	custom Processor stores

getKeyValueStore transparently unwraps DSL-created timestamped stores, so it returns the raw value V (matching Kafka Streams behaviour). The state-store name is the one you set via Materialized.as("...") or the store supplier.

val builder = StreamsBuilder()
builder
    .stream<String, String>("input-topic")
    .groupByKey()
    .count(Materialized.`as`<String, Long, StateStore>("count-store"))
    .toStream()
    .to("output-topic")

val driver = TopologyTestDriver.fromBuilder(builder, TopologyTestDriver.STRING_CONFIG)
val input = driver.createInputTopic("input-topic", Serdes.String(), Serdes.String())

input.pipeInput("a", "x")
input.pipeInput("a", "y")
input.pipeInput("b", "z")

val store = driver.getKeyValueStore<String, Long>("count-store")
assertThat(store.get("a")).isEqualTo(2L)
assertThat(store.get("b")).isEqualTo(1L)
assertThat(store.get("missing")).isNull()

StreamsBuilder builder = new StreamsBuilder();
builder.<String, String>stream("input-topic")
    .groupByKey()
    .count(Materialized.<String, Long, StateStore>as("count-store"))
    .toStream()
    .to("output-topic");

TopologyTestDriver driver = TopologyTestDriver.fromBuilder(builder, TopologyTestDriver.STRING_CONFIG);
TestInputTopic<String, String> input =
    driver.createInputTopic("input-topic", Serdes.String(), Serdes.String());

input.pipeInput("a", "x");
input.pipeInput("a", "y");
input.pipeInput("b", "z");

KeyValueStore<String, Long> store = driver.getKeyValueStore("count-store");
assertThat(store.get("a")).isEqualTo(2L);
assertThat(store.get("b")).isEqualTo(1L);
assertThat(store.get("missing")).isNull();

The test driver does not write changelog topics — it runs without a broker, so state writes use a no-op changelog. Tests cover processing and state logic, not state backup / restore. To exercise changelog and recovery behaviour, use a broker-backed integration test (the example apps pair each ...AppTest with a ...IntegrationTest).

Flushing caches with commitBarrier()

State stores with caching enabled (via Materialized.withCachingEnabled()) suppress intermediate updates and emit only the final value per key on commit. In the runtime that happens on a commit barrier; in the test driver you trigger it explicitly with commitBarrier(). advanceWallClockTime(...) also triggers one. If a caching-enabled topology emits nothing after pipeInput, call commitBarrier() (or advance wall-clock time) to flush.

input.pipeInput("a", "x")
input.pipeInput("a", "y")

// Nothing emitted yet — caching holds intermediate updates
assertThat(output.isEmpty()).isTrue()

driver.commitBarrier()                 // flush cached emissions

assertThat(output.readKeyValuesToMap()).containsEntry("a", 2L)

input.pipeInput("a", "x");
input.pipeInput("a", "y");

// Nothing emitted yet — caching holds intermediate updates
assertThat(output.isEmpty()).isTrue();

driver.commitBarrier();                // flush cached emissions

assertThat(output.readKeyValuesToMap()).containsEntry("a", 2L);

Triggering scheduled sources

A scheduled source generates records on a clock rather than from Kafka. You drive it two ways:

triggerScheduledSource(name) fires the emitter once, immediately, without advancing time — handy for asserting the emitter logic in isolation.
advanceWallClockTime(...) (for WALL_CLOCK_TIME / cron sources) or advanceWatermark(...) (for STREAM_TIME sources) fires every scheduled execution due up to the new time.

The default name of the first builder.scheduled(...) source is scheduled-source-0; pass a Named to the operator to set your own. In Kotlin the key serde is optional (defaults to null); in Java, pass it explicitly since Java has no default arguments.

val builder = StreamsBuilder()
builder
    .scheduled<String, String>(
        interval = Duration.ofSeconds(10),
        type = PunctuationType.WALL_CLOCK_TIME,
        emitter = { ctx -> ctx.forward("hb", "heartbeat-${ctx.currentWallClockTime()}") },
    )
    .to("output", Produced.with(Serdes.String(), Serdes.String()))

val driver = TopologyTestDriver.fromBuilder(builder)
val output = driver.createOutputTopic<String, String>("output", Serdes.String(), Serdes.String())

// Fire once, no time advance
driver.triggerScheduledSource("scheduled-source-0")
assertThat(output.readRecord()?.key).isEqualTo("hb")

// Or fire by advancing the wall clock past the interval
driver.advanceWallClockTime(Duration.ofSeconds(10))
assertThat(output.isEmpty()).isFalse()

StreamsBuilder builder = new StreamsBuilder();
builder.<String, String>scheduled(
        Duration.ofSeconds(10),
        PunctuationType.WALL_CLOCK_TIME,
        Serdes.String(),   // key serde (Java has no default args)
        ctx -> ctx.forward("hb", "heartbeat-" + ctx.currentWallClockTime()))
    .to("output", Produced.with(Serdes.String(), Serdes.String()));

TopologyTestDriver driver = TopologyTestDriver.fromBuilder(builder);
TestOutputTopic<String, String> output =
    driver.createOutputTopic("output", Serdes.String(), Serdes.String());

// Fire once, no time advance
driver.triggerScheduledSource("scheduled-source-0");
assertThat(output.readRecord().getKey()).isEqualTo("hb");

// Or fire by advancing the wall clock past the interval
driver.advanceWallClockTime(Duration.ofSeconds(10));
assertThat(output.isEmpty()).isFalse();

Config-driven tests (StoatFlowTestDriver)

StoatFlowTestDriver.fromConfig(...) is the test-time mirror of the production StoatFlowRuntime.fromConfig(...) (see Your first app). It loads application.yaml from the classpath, maps it into a StreamsConfig, builds your topology with the configured default serdes, and returns a ready TopologyTestDriver. No broker, no HTTP server — just the topology under the same configuration your app runs with.

Configuration is resolved from the classpath, so a src/test/resources/application.yaml shadows the main one — letting tests pin a deterministic config. Environment variables with the STOATFLOW_ prefix take highest priority; the optional configOverrides lambda overrides programmatically above everything else.

The topology-builder callback receives both the StreamsBuilder and the loaded StreamsConfig, so a topology that needs config values (e.g. config.schemaRegistryUrl) can read them.

import io.stoatflow.testruntime.StoatFlowTestDriver
import io.stoatflow.testutils.TopologyTestDriver

private fun buildDriver(): TopologyTestDriver =
    StoatFlowTestDriver.fromConfig(
        topologyBuilder = { builder, config -> MyApp.buildTopology(builder, config) },
        // optional: configOverrides = { /* StreamsConfig.Builder receiver */ },
    )

import io.stoatflow.testruntime.StoatFlowTestDriver;
import io.stoatflow.testutils.TopologyTestDriver;

static TopologyTestDriver buildDriver() {
    return StoatFlowTestDriver.fromConfig(
        (builder, config) -> MyApp.buildTopology(builder),
        null   // optional Consumer<StreamsConfig.Builder> for overrides
    );
}

From there it is the same TopologyTestDriver API — createInputTopic, createOutputTopic, advanceWatermark, getKeyValueStore, and so on. This is the pattern the example apps use; see examples/stock-tick-filter (Kotlin) and examples/ecommerce-daily-customer-behaviour / examples/news-feed-subscription-processor (Java) for full end-to-end topology tests, each paired with a broker-backed integration test.

Porting Kafka Streams tests

If you're migrating a Kafka Streams test suite, you don't have to rewrite it around fromBuilder. The driver also carries the KS-shaped surface, so an existing KS test ports with the same import swap as the topology it exercises:

KS constructors — TopologyTestDriver(topology), (topology, Properties), (topology, Instant), and (topology, Properties, Instant). The Properties overloads accept KS-keyed config (application.id, default serde class names, …) through the same adapter the engine uses. Typed (topology, StreamsConfig[, Instant]) forms exist too.
fromTopology(topology[, config][, initialWallClockTime]) — for a Topology assembled with the Processor API (addSource / addProcessor / addSink) rather than a StreamsBuilder.
getAllStateStores() — every state store keyed by name, matching KS.
producedTopicNames() — the set of topics the driver has produced records to, matching KS.

val props = Properties()
props[StreamsConfig.APPLICATION_ID_CONFIG] = "uppercase-test"
props[StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG] = Serdes.String()::class.java
props[StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG] = Serdes.String()::class.java

// The KS constructor idiom, unchanged
val driver = TopologyTestDriver(builder.build(), props)

input.pipeInput("k1", "hello")

driver.producedTopicNames()   // setOf("output-topic")
driver.getAllStateStores()    // all stores by name

var props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "uppercase-test");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

// The KS constructor idiom, unchanged
var driver = new TopologyTestDriver(builder.build(), props);

input.pipeInput("k1", "hello");

driver.producedTopicNames();   // Set.of("output-topic")
driver.getAllStateStores();    // all stores by name

For new StoatFlow tests, prefer fromBuilder (above) — it skips the builder.build() step and the Properties indirection. The KS-shaped surface exists so ported suites keep passing as-is; the KS compatibility matrix tracks the full test-utils parity.

Next steps

Streams builder — building the topology you're testing.
State stores — the store types the accessors above return.
Windowing and Event time and watermarks — the time model behind advanceWatermark.
Scheduled sources — clock-driven sources and how to test them.

State stores

Use the Stores factory and Materialized to configure persistent and in-memory key-value, window, session, and versioned stores — and read them back via interactive queries.

Sources and sinks

Read records into a topology with builder.stream / table / globalTable and Consumed, and write them back out with to() and Produced — serdes, offset reset, watermarks, partitioners, and the fan-out rule.