KStream and KTable operations

Stateless stream transforms (map, filter, flatMap, selectKey, branch, merge) and KTable basics — with the rules for when a key change forces records onto a different lane.

KStream<K, V> is an unbounded stream of records; KTable<K, V> is a changelog table holding one value per key. This page covers the stateless KStream transforms and the basic KTable operations. Aggregations, joins, windowing, and the Processor API have their own pages.

Both types are created through the StreamsBuilder, never constructed directly. The operators are Kafka Streams compatible — Java code written against the KS DSL compiles against StoatFlow with no changes (see the Kafka Streams compatibility matrix).

Two kinds of operator: value-preserving vs key-changing

The single most important distinction in this API is whether an operator changes the key. StoatFlow routes every record to a processing lane by key affinity — same key, same lane, in order. An operator that changes the key forces the record to be re-routed to whichever lane owns the new key. (This is StoatFlow's in-memory equivalent of a Kafka Streams repartition topic — no broker round-trip. See Lanes and parallelism and Architecture.)

Operator	Changes key?	Effect
`mapValues`	No	Stays on the same lane
`filter` / `filterNot`	No	Stays on the same lane
`flatMapValues`	No	All output records keep the input key
`peek`	No	Pure side effect
`merge`	No	Records keep their original key affinity
`split`	No	Routing only — keys unchanged
`map`	Yes	Re-routed by new key
`flatMap`	Yes	Each output re-routed by its key
`selectKey`	Yes	Re-routed by new key
`groupBy`	Yes	Re-keys, then groups

Prefer the value-only operators (mapValues, flatMapValues) when you don't need to change the key — they avoid re-routing entirely.

Naming operators

Every operator accepts an optional trailing Named parameter. StoatFlow uses these stable names for the topology graph, metrics, and (for stateful operators) state-store identity. In Kotlin as is a soft keyword, so escape it with backticks: Named.`as`("uppercase"). In Java it's a plain method: Named.as("uppercase").

import io.stoatflow.core.topology.Named

Naming is optional but recommended — unnamed nodes get auto-generated names that change as you edit the topology.

Value transforms — `mapValues`

mapValues transforms each value and keeps the key. There's a value-only form and a key-aware form (the key is read-only — it is not changed).

import io.stoatflow.core.topology.Named

// value-only
val upper = stream.mapValues({ v -> v.uppercase() }, Named.`as`("uppercase"))

// key-aware (key is read-only)
val tagged = stream.mapValues({ k, v -> "$k:$v" }, Named.`as`("tag-with-key"))

import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.ValueMapper;
import io.stoatflow.core.topology.ValueMapperWithKey;

// value-only
KStream<String, String> upper =
    stream.mapValues((ValueMapper<String, String>) v -> v.toUpperCase(), Named.as("uppercase"));

// key-aware (key is read-only)
KStream<String, String> tagged =
    stream.mapValues((ValueMapperWithKey<String, String, String>) (k, v) -> k + ":" + v,
        Named.as("tag-with-key"));

Values may be null (a tombstone). Your mapper, predicate, and action functions must handle null values — StoatFlow does not filter them out for you.

Key + value transform — `map`

map returns a KeyValue<KR, VR>, replacing both key and value. This is a key-changing operation — the output is re-routed to the lane that owns the new key.

import io.stoatflow.core.state.KeyValue
import io.stoatflow.core.topology.Named

val rekeyed = stream.map(
    { key, value -> KeyValue(value.substring(0, 3), "$key:$value") },
    Named.`as`("rekey-by-prefix"),
)

import io.stoatflow.core.state.KeyValue;
import io.stoatflow.core.topology.KeyValueMapper;
import io.stoatflow.core.topology.Named;

KStream<String, String> rekeyed = stream.map(
    (KeyValueMapper<Long, String, KeyValue<String, String>>) (key, value) ->
        new KeyValue<>(value.substring(0, 3), key + ":" + value),
    Named.as("rekey-by-prefix"));

Filtering — `filter` and `filterNot`

filter keeps records that match the predicate; filterNot keeps records that don't. Neither changes the key.

import io.stoatflow.core.topology.Named

val longEnough = stream.filter({ _, v -> v.length > 5 }, Named.`as`("keep-long"))
val notEmpty = stream.filterNot({ _, v -> v.isEmpty() }, Named.`as`("drop-empty"))

import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.Predicate;

KStream<Long, String> longEnough =
    stream.filter((Predicate<Long, String>) (k, v) -> v.length() > 5, Named.as("keep-long"));
KStream<Long, String> notEmpty =
    stream.filterNot((Predicate<Long, String>) (k, v) -> v.isEmpty(), Named.as("drop-empty"));

One-to-many — `flatMapValues` and `flatMap`

flatMapValues turns each value into zero or more values, all keeping the input key — value-only, so no re-routing. There's also a key-aware form that reads (but does not change) the key.

import io.stoatflow.core.topology.Named

val whitespace = "\\s+".toRegex()
val words = lines.flatMapValues(
    { line -> line.lowercase().split(whitespace).filter { it.isNotBlank() } },
    Named.`as`("split-words"),
)

import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.ValueMapper;

import java.util.Arrays;
import java.util.stream.Collectors;

KStream<String, String> words = lines.flatMapValues(
    (ValueMapper<String, Iterable<String>>) line ->
        Arrays.stream(line.toLowerCase().split("\\s+"))
            .filter(w -> !w.isBlank())
            .collect(Collectors.toList()),
    Named.as("split-words"));

flatMap returns an Iterable<KeyValue<KR, VR>> — each output record carries its own key, so this is a key-changing operation and each output is re-routed independently.

import io.stoatflow.core.state.KeyValue
import io.stoatflow.core.topology.Named

val perTag = stream.flatMap(
    { _, value -> value.tags.map { tag -> KeyValue(tag, value.id) } },
    Named.`as`("explode-tags"),
)

import io.stoatflow.core.state.KeyValue;
import io.stoatflow.core.topology.KeyValueMapper;
import io.stoatflow.core.topology.Named;

import java.util.stream.Collectors;

KStream<String, String> perTag = stream.flatMap(
    (KeyValueMapper<String, Doc, Iterable<KeyValue<String, String>>>) (k, value) ->
        value.tags().stream()
            .map(tag -> new KeyValue<>(tag, value.id()))
            .collect(Collectors.toList()),
    Named.as("explode-tags"));

Re-keying — `selectKey`

selectKey replaces only the key, leaving the value alone. It is a key-changing operation — use it before groupByKey or a join when you need to key the stream on a value field.

import io.stoatflow.core.topology.Named

val byCustomer = orders.selectKey({ _, order -> order.customerId }, Named.`as`("key-by-customer"))

import io.stoatflow.core.topology.KeyValueMapper;
import io.stoatflow.core.topology.Named;

KStream<String, Order> byCustomer =
    orders.selectKey((KeyValueMapper<String, Order, String>) (k, order) -> order.customerId(),
        Named.as("key-by-customer"));

Side effects — `peek`

peek runs an action for each record (logging, counters, debugging) and forwards the record unchanged. The key is never modified. (forEach is the terminal variant — it runs the action but returns nothing, ending the chain.)

import io.stoatflow.core.topology.Named

val observed = stream.peek({ k, v -> println("$k => $v") }, Named.`as`("log"))

import io.stoatflow.core.topology.ForeachAction;
import io.stoatflow.core.topology.Named;

KStream<Long, String> observed =
    stream.peek((ForeachAction<Long, String>) (k, v) -> System.out.println(k + " => " + v),
        Named.as("log"));

The Java overloads take a ForeachAction, so void lambdas work directly — no return Unit.INSTANCE.

Combining streams — `merge`

merge combines two streams of the same key/value types into one. It is not a key-changing operation: records keep their original key affinity. Both streams must come from the same StreamsBuilder.

import io.stoatflow.core.topology.Named

val combined = streamA.merge(streamB, Named.`as`("merge-a-b"))

import io.stoatflow.core.topology.Named;

KStream<String, String> combined = streamA.merge(streamB, Named.as("merge-a-b"));

Branching — `split`

split routes each record to exactly one branch using first-match semantics: predicates are evaluated in order, and the first match wins. split() returns a BranchedKStream; the terminal defaultBranch() / noDefaultBranch() call returns a map of branch name to KStream. Branch names are the split name plus the Branched suffix — e.g. split name router plus suffix -high gives key router-high. split does not change the key.

Finish the chain with defaultBranch() (unmatched records go to a default branch), defaultBranch(Branched.as("-suffix")) (named default), or noDefaultBranch() (unmatched records are dropped).

import io.stoatflow.core.topology.Branched
import io.stoatflow.core.topology.Named

val branches: Map<String, KStream<Long, String>> = stream
    .split(Named.`as`("router"))
    .branch({ _, v -> v.length > 20 }, Branched.`as`("-high"))
    .branch({ _, v -> v.length > 10 }, Branched.`as`("-medium"))
    .defaultBranch(Branched.`as`("-low"))

branches["router-high"]!!.to("high-topic")
branches["router-medium"]!!.to("medium-topic")
branches["router-low"]!!.to("low-topic")

import io.stoatflow.core.topology.Branched;
import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.Predicate;

import java.util.Map;

Map<String, KStream<Long, String>> branches = stream
    .split(Named.as("router"))
    .branch((Predicate<Long, String>) (k, v) -> v.length() > 20, Branched.as("-high"))
    .branch((Predicate<Long, String>) (k, v) -> v.length() > 10, Branched.as("-medium"))
    .defaultBranch(Branched.as("-low"));

branches.get("router-high").to("high-topic");
branches.get("router-medium").to("medium-topic");
branches.get("router-low").to("low-topic");

Branched can also inline a branch transform or side effect instead of returning the stream in the map: Branched.withFunction(s -> s.mapValues(...), "-medium") applies a transform and stores the result; Branched.withConsumer(s -> s.to("topic"), "-low") applies a side effect and stores the original branch stream.

KTable basics

A KTable represents the latest value per key as a changelog. You get one from StreamsBuilder.table(...), from a stream-to-table conversion (KStream.toTable), or as the output of an aggregation. The stateless KTable operators mirror the stream ones but operate on the table's changelog.

`mapValues` and `filter`

KTable.mapValues transforms values (value-only and key-aware forms). KTable.filter / filterNot keep or drop entries by predicate. Without a Materialized argument these are derived, on-the-fly views over the source table's state and keep no store of their own.

A materialized KTable.filter differs semantically from the non-materialized form: when an entry stops matching, a materialized filter writes a tombstone (null) to its store and forwards it downstream, whereas the non-materialized filter simply doesn't forward the record. Pass a Materialized argument to opt into the store. See State stores.

import io.stoatflow.core.topology.Named

val upper = table
    .mapValues({ v -> v.uppercase() }, Named.`as`("table-upper"))
    .filter({ _, v -> v.isNotEmpty() }, Named.`as`("table-non-empty"))

import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.Predicate;
import io.stoatflow.core.topology.ValueMapper;

KTable<String, String> upper = table
    .mapValues((ValueMapper<String, String>) v -> v.toUpperCase(), Named.as("table-upper"))
    .filter((Predicate<String, String>) (k, v) -> !v.isEmpty(), Named.as("table-non-empty"));

`toStream` — table back to stream

KTable.toStream converts the table's changelog into a KStream: every update to a key becomes a record. This is how you write a table's results to a topic. (count, reduce, and aggregate produce a KTable, so toStream is the usual bridge to a sink — see Aggregations.)

import io.stoatflow.core.topology.Named

table
    .mapValues({ v -> v.uppercase() }, Named.`as`("table-upper"))
    .toStream(Named.`as`("to-stream"))
    .to("output-topic")

import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.ValueMapper;

table
    .mapValues((ValueMapper<String, String>) v -> v.toUpperCase(), Named.as("table-upper"))
    .toStream(Named.as("to-stream"))
    .to("output-topic");

There's a key-changing toStream(keyMapper, named) overload that derives a new key from each entry — like selectKey, it re-routes the resulting records to the lane that owns the new key.

`toTable` — stream to table

KStream.toTable interprets a stream as a changelog: each record updates the key's value in the resulting table, and a null value is a tombstone that deletes the key. It preserves the key (no re-routing). Pass a Materialized to control the backing store and its name; without one StoatFlow auto-generates a store name.

import io.stoatflow.core.state.StateStore
import io.stoatflow.core.topology.Materialized
import org.apache.kafka.common.serialization.Serdes

val latest = stream.toTable(
    Materialized.`as`<String, String, StateStore>("latest-by-key")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.String()),
)

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.topology.Materialized;
import org.apache.kafka.common.serialization.Serdes;

KTable<String, String> latest = stream.toTable(
    Materialized.<String, String, StateStore>as("latest-by-key")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.String()));

Putting it together

A complete stateless pipeline: read a topic, re-key, filter, uppercase, and write the result. (Adapted from the map-filter example.)

import io.stoatflow.core.state.KeyValue
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("input-topic", Consumed.`as`<Long, String>("source").withKeySerde(Serdes.Long()))
        .map({ key, value -> KeyValue(value.substring(0, 3), "$key:$value") }, Named.`as`("rekey"))
        .filter({ _, value -> value.length > 5 }, Named.`as`("keep-long"))
        .mapValues({ value -> value.uppercase() }, Named.`as`("uppercase"))
        .to(
            "output-topic",
            Produced.`as`<String, String>("sink")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.String()),
        )
}

import io.stoatflow.core.state.KeyValue;
import io.stoatflow.core.topology.Consumed;
import io.stoatflow.core.topology.KStream;
import io.stoatflow.core.topology.KeyValueMapper;
import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.Predicate;
import io.stoatflow.core.topology.Produced;
import io.stoatflow.core.topology.StreamsBuilder;
import io.stoatflow.core.topology.ValueMapper;
import org.apache.kafka.common.serialization.Serdes;

void buildTopology(StreamsBuilder builder) {
    KStream<Long, String> input =
        builder.stream("input-topic", Consumed.<Long, String>as("source").withKeySerde(Serdes.Long()));
    input
        .map((KeyValueMapper<Long, String, KeyValue<String, String>>) (key, value) ->
                new KeyValue<>(value.substring(0, 3), key + ":" + value),
            Named.as("rekey"))
        .filter((Predicate<String, String>) (k, value) -> value.length() > 5, Named.as("keep-long"))
        .mapValues((ValueMapper<String, String>) String::toUpperCase, Named.as("uppercase"))
        .to(
            "output-topic",
            Produced.<String, String>as("sink")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.String()));
}

Next steps

Aggregations — groupByKey, groupBy, count, reduce, aggregate.
Joins — stream-table, stream-stream, and table-table joins.
Serdes — Consumed, Produced, and serde resolution.
Lanes and parallelism — how key affinity and re-routing work.
Testing — verify these operators with the in-memory test driver.

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.

Aggregations

Group records by key and fold them into running totals — count, reduce, aggregate on grouped streams and grouped tables, with the result backed by a state store.