Building topologies

A topology is the processing graph your application runs: source topics in, operators in the middle, sink topics out. You describe it once with StreamsBuilder, hand the builder to the runtime, and the engine executes it. The DSL is Kafka Streams compatible — if you know stream(), mapValues(), groupBy(), and to(), you already know how to write one.

This page is the orientation for the section: the entry point, the stream abstractions at a glance, and the one rule that trips up people coming from Kafka Streams — the fan-out rule. Each operator family has its own page; links are at the bottom.

StreamsBuilder — the entry point

Everything starts with a StreamsBuilder. Its source methods (stream, table, globalTable, scheduled) return DSL objects you chain operators onto; terminal operations (to, forEach, print) end a branch and return nothing. You never construct KStream or KTable directly — the builder hands them to you.

A source method takes a topic name and an optional Consumed for serdes, watermark strategy, and offset-reset policy. Every operator takes an optional Named so its node gets a stable identity in the topology graph, metrics, and state-store names. The convention across the examples is to name each operator explicitly.

Here is a minimal end-to-end topology — read, transform, filter, write:

import io.stoatflow.core.topology.Consumed
import io.stoatflow.core.topology.Named
import io.stoatflow.core.topology.Produced
import io.stoatflow.core.topology.StreamsBuilder
import org.apache.kafka.common.serialization.Serdes

fun buildTopology(builder: StreamsBuilder) {
    builder
        .stream<String, String>("input-topic", Consumed.`as`("source"))
        .mapValues({ value -> value.uppercase() }, Named.`as`("upper"))
        .filter({ _, value -> value.length > 5 }, Named.`as`("longer-than-5"))
        .to(
            "output-topic",
            Produced.`as`<String, String>("sink")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.String()),
        )
}

import io.stoatflow.core.topology.Consumed;
import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.Produced;
import io.stoatflow.core.topology.StreamsBuilder;
import org.apache.kafka.common.serialization.Serdes;

void buildTopology(StreamsBuilder builder) {
    builder
        .<String, String>stream("input-topic", Consumed.as("source"))
        .mapValues(value -> value.toUpperCase(), Named.as("upper"))
        .filter((key, value) -> value.length() > 5, Named.as("longer-than-5"))
        .to(
            "output-topic",
            Produced.<String, String>as("sink")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.String()));
}

When you're done describing the graph, the builder is what you pass to the runtime — StoatFlowRuntime.fromConfig(topologyBuilder = { buildTopology(it) }, ...). The runtime calls build() for you and validates the topology before it starts. (If you're embedding the core engine directly rather than using the runtime wrapper, you call builder.build() yourself.)

The stream abstractions at a glance

The DSL has two foundational types and several specialised ones you reach by grouping or windowing:

StoatFlow also adds scheduled sources — a source that emits records on an interval or cron schedule instead of consuming a topic. It returns a KStream and flows through the topology like any other source. See Scheduled sources.

The mental model is the same as Kafka Streams: a KStream is a stream of events; a KTable is the latest-value-per-key view of a changelog. KStream.toStream() doesn't exist (it's already a stream), but KTable.toStream() turns a table back into its change events, and KStream.toTable() materialises a stream as a table. For the full operator-by-operator parity status against Kafka Streams, see the Kafka Streams compatibility matrix.

The fan-out rule

This is the one rule worth internalising before you write anything non-trivial.

To send one input stream down multiple processing paths — say, one branch to topic A and another to topic B — you reuse the same KStream reference for each branch. You do not call builder.stream() twice on the same topic. Subscribing the same topic from two source nodes is rejected when the topology is built, with a TopologyValidationException.

builder.stream<String, String>("orders").filter(...).to("a")   // source node 1
builder.stream<String, String>("orders").mapValues(...).to("b") // source node 2 — REJECTED at build()

Instead, hold the source KStream in a variable and branch off it. Each operator chain that starts from the shared reference is an independent path through the topology; they all share the single source subscription:

import io.stoatflow.core.topology.Consumed
import io.stoatflow.core.topology.KStream
import io.stoatflow.core.topology.Named
import io.stoatflow.core.topology.Produced
import io.stoatflow.core.topology.StreamsBuilder

fun buildTopology(builder: StreamsBuilder) {
    // Read the topic ONCE, keep the reference.
    val orders: KStream<String, String> =
        builder.stream("orders", Consumed.`as`("source"))

    // Branch 1: filter, write to "large-orders".
    orders
        .filter({ _, v -> v.length > 100 }, Named.`as`("only-large"))
        .to("large-orders", Produced.`as`("large-sink"))

    // Branch 2: transform the SAME source, write to "orders-upper".
    orders
        .mapValues({ v -> v.uppercase() }, Named.`as`("upper"))
        .to("orders-upper", Produced.`as`("upper-sink"))
}

import io.stoatflow.core.topology.Consumed;
import io.stoatflow.core.topology.KStream;
import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.Produced;
import io.stoatflow.core.topology.StreamsBuilder;

void buildTopology(StreamsBuilder builder) {
    // Read the topic ONCE, keep the reference.
    KStream<String, String> orders =
        builder.stream("orders", Consumed.as("source"));

    // Branch 1: filter, write to "large-orders".
    orders
        .filter((key, v) -> v.length() > 100, Named.as("only-large"))
        .to("large-orders", Produced.as("large-sink"));

    // Branch 2: transform the SAME source, write to "orders-upper".
    orders
        .mapValues(v -> v.toUpperCase(), Named.as("upper"))
        .to("orders-upper", Produced.as("upper-sink"));
}

The same rule applies to every topic-backed source: reuse the returned reference from table() and globalTable() rather than calling them twice for the same topic. scheduled() is not topic-backed — each call defines its own independent source (there is no shared Kafka topic to conflict over), so two scheduled() calls are never rejected. To branch a scheduled source's output, reuse its returned KStream reference the same way you would any other source.

Where to go next

Each operator family has a dedicated how-to:

• StreamsBuilder — sources, sinks, global/read-only stores, and build() in depth.
• KStream and KTable — map, filter, flatMap, selectKey, merge, split, table operations, and stream↔table conversion.
• Aggregations — count, reduce, aggregate, grouping, and cogroup.
• Windowing — tumbling, hopping, sliding, and session windows.
• Joins — stream-stream, stream-table, and table-table (incl. foreign-key) joins.
• Processor API — custom Processor / FixedKeyProcessor, state-store access, timers, and punctuators.
• Scheduled sources — interval- and cron-driven sources (a StoatFlow extension).
• Serdes — configuring serialization for keys and values.
• State stores — store types, materialization, and the Materialized config.
• Error handling & DLQ — deserialization and processing exception handlers.
• Testing — the in-memory TopologyTestDriver, no broker required.

For the engine that runs the topology you build here — lanes, commit barriers, state durability — see Architecture.