StoatFlow

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>")
}

The current version is 1.0.0-rc.4. 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.

The driver also performs the engine's boundary key serialization: wherever a record (or a window-close / suppress / punctuator emission) crosses a sub-topology boundary — where a re-keyed record reaches an aggregation, a join, a materialised table, or an explicit repartition() — its key is serialized with the same boundary key serde the engine resolves, and an unresolvable boundary raises the same LaneKeySerializationException your application would hit in production. A topology that would die on its first record can no longer run green in a unit test.

The driver compiles the topology with the same topology.sub-topology-split and topology.processor-api-key-affinity settings as the engine, so the boundaries it exercises are the ones your application will actually have — and topology-validation rules set to error fail the driver too, with no broker. A re-key feeding a store-connected Processor API node is refused in your unit tests rather than in production (Key affinity after a re-key).

The driver runs every store in memory, but the exactly-once store checks judge the store you declared, not its in-memory stand-in. A table a join reads from another lane that you materialised with Stores.persistentVersionedKeyValueStore(...) passes in the driver, which runs its in-memory twin, as it does in production. One you materialised with Stores.inMemoryVersionedKeyValueStore(...) or Stores.lruMap(...) is refused under exactly_once (the default) in both (Joins under exactly-once). To test a primary-key or GlobalKTable join over a versioned table, declare the RocksDB versioned store.

One thing the driver cannot show you: it is single-lane, so any concurrency property — including the lane-affinity trade-off above — is invisible to it by construction. Those need a real broker.

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()
    }
}
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. Several drivers may share one StreamsBuilder or Topology, though — a shared test fixture, two drivers compared in one test, or a driver over the builder of a running application: each driver builds its own processors and state stores from it, as in Kafka Streams. A driver given no config compiles the settings the topology was built with, whichever run compiled it before. Two things stay shared, because the topology holds them as your own objects: a scheduled source's emitter — every run invokes the same object, so keep it stateless or give each driver its own builder — and the stream of a Printed.toFile (as in Kafka Streams).

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. A record piped without a timestamp — pipeInput(key, value), or a TestRecord whose timestamp is unset (its default, as in Kafka Streams) — takes the topic's current time. That time starts at the driver's wall clock when the topic is created and moves only through the topic itself — its withAutoAdvance step, withTimestamp, advanceTime or reset(); a later advanceWallClockTime does not move it, as in Kafka Streams.

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
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
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 — see commitBarrier(). It is also the clock your code reads: currentSystemTimeMs() on every processor, timer, punctuator and scheduled-source context, and TimerService.currentProcessingTime(), return the driver's wall clock, as in Kafka Streams' driver. Pin its start with initialWallClockTime for exact assertions.

The driver initialises the topology's processors when it is built, as Kafka Streams' driver does: each Processor API processor's init() runs in the constructor, so a WALL_CLOCK_TIME punctuator scheduled there fires on advanceWallClockTime before any record is piped, and an init() that throws (an undeclared state store, say) fails the constructor.

A punctuator that throws — a forward to a child name that does not exist included — is handled as in production: it goes to the processing exception handler, so with the default handler the advanceWatermark / advanceWallClockTime call throws, and with a handler that returns CONTINUE the driver logs it and carries on.

A timer whose onTimer throws, or whose key cannot be deserialized when it fires, is handled the same way: the handler gets the synthetic __timer context and the timer's key (the raw key bytes if decoding failed), and its dead letters land on the dead-letter topic you can read with createOutputTopic. On FAIL the advance throws once all of the timers that were due have fired, so one failing timer does not hide the others.

The driver also checks that a Processor API processor's timer key serde puts a key's timers on its records' lane in production (Timer key serde). It runs every key on one lane, but judges a split over the processor's sub-topology lane count — its numberOfLanes override, else num.stream.threads, else a fixed reference count of 16 (never the build machine's cores, so a verdict is the same on every machine). Set num.stream.threads in your test Properties to your production value to judge over exactly that. A repartition() StreamPartitioner is applied per record, as in production: a record it lanes itself is not checked, one it hands back to the key hash (Optional.empty()) is, and one it drops never reaches the processor. Where production samples, the driver checks every piped record whose key it can compare: one with a key, not re-keyed before the processor, that the timer key serde can serialize. That sampled verdict follows timer.lane-affinity.sampled-check — warn by default, so a split is logged, not thrown; set it to error to make a wrongly configured key serde fail the test. A timer registered for the record's own key that picks another lane follows timer.lane-affinity.guard — error by default, so it throws TimerLaneAffinityException.

Set test.driver.commit-per-pipe=false in your config Properties to opt out of the per-pipe parity: pipes then neither commit the epoch nor advance event time, and you control both explicitly via commitBarrier() / advanceWatermark() (where a backward request throws, the legacy manual-control mode).
The driver's clock is not production's. The driver's event time is the maximum timestamp it has seen — the source WatermarkStrategy's generator never runs, there is one partition and no out-of-orderness lag, and final emissions run against that raw value synchronously. Production keeps the watermark as the minimum across partitions minus the bound (10 s by default) and emits a final result at a commit barrier up to two commit intervals later. A window-close or join-null-side test that passes here therefore emits later on a real broker; both count the same records. To make the two agree, use forMonotonousTimestamps() or max-out-of-orderness-ms: 0 in the test and drive event time with advanceWatermark; put emission-timing assertions in a broker-backed integration test. The full reasoning is on Event time and watermarks. The driver has no partition buffers either, so a record stamped in the future is processed as soon as it is piped — it is never held behind older records, and stoatflow.watermark.preview.partitions has no driver counterpart. For the same reason the driver does not model watermark alignment: withWatermarkAlignment(...) holds nothing in a test, so a hold that would keep a real application waiting never shows here. The two alignment topology rules — alignment-idleness-off and alignment-mixed-watermark-bounds — do run, because they are checked when the topology is compiled; a test fails to build where the application would.Pipe every topic in event time, tables included. Because the driver's watermark is the largest timestamp piped into any topic, and the watermark decides which records are late, a table record piped without a timestamp (it gets the wall clock) moves it past every window of older test data — and those records are then dropped as late. Give pipeInput the same event-time timestamps for a table as for the streams around it.The driver also runs everything as one sub-topology. In production, a windowed operation fed by another window's final results across a re-key closes about two commit intervals later per step, so the driver shows those finals earlier than a broker does.

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

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.

AccessorReturnsFor
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 : StateStorecustom Processor stores

getKeyValueStore transparently unwraps DSL-created timestamped stores, so it returns the raw value V (matching Kafka Streams behaviour). The same holds for a store materialised with withRecordHeaders(): getKeyValueStore, getWindowStore and getSessionStore return the plain value, and getTimestampedKeyValueStore / getTimestampedWindowStore return ValueAndTimestamp with the headers dropped. Writing through them stores the entry with empty headers — put, and also compute and merge — as in Kafka Streams, where a driver write has no record context. Set test.driver.keep-existing-headers=true to have every such write to an existing entry keep that entry's headers instead (a new entry still gets empty headers). The one exception is a window store that retains duplicates — every join buffer, or one built with retainDuplicates = true — where a put appends a record rather than updating one: there the flag has no effect, and the driver logs a warning saying so. getVersionedKeyValueStore returns plain VersionedRecords over a headers-aware versioned store in the same way. Use the *WithHeaders accessors to read or set the headers. 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()
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).

What commitBarrier() does

commitBarrier() commits the open epoch, as a commit barrier does in the runtime: state stores write their changes and send their changelogs, and barrier-gated suppress buffers (Suppressed.untilBarrier(), or every aggregation under stoatflow.emit.mode: on-barrier) emit their consolidated records. advanceWallClockTime(...) also triggers one, and by default every pipeInput does.

It does not release held-back output, because none is held back. Caching (Materialized.withCachingEnabled(), the default) batches store and changelog writes; it never changes what is emitted downstream. Every update is forwarded the moment it happens, with the record's timestamp and headers, exactly as in production. This is also a difference from Kafka Streams, whose record cache consolidates downstream emissions. To get one record per key per epoch, use untilBarrier() or emit.mode: on-barrier, and drive the epochs with commitBarrier():

// config: test.driver.commit-per-pipe=false, so the epoch spans several pipes
builder.stream<String, String>("in").groupByKey().count()
    .suppress(Suppressed.untilBarrier())
    .toStream().to("out")

input.pipeInput("a", "x")
input.pipeInput("a", "y")
assertThat(output.isEmpty()).isTrue()   // the suppress buffer holds the epoch

driver.commitBarrier()                 // the epoch closes: one consolidated record
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()

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 */ },
    )

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

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

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.