Lanes and parallelism

A lane is StoatFlow's unit of concurrent processing inside the single JVM. Records are routed to lanes by their key: the same key always lands on the same lane, so the topology sees that key's events in arrival order; different keys run on different lanes in parallel. This page explains what that buys you and how to think about lane count. For the runtime mechanism behind it, see Architecture.

Key affinity: same key, same lane

When a record arrives, the engine inspects its key and consistently routes it to one lane — the same key maps to the same lane every time. Because one lane processes one key's events sequentially, in the order Kafka delivered them, per-key ordering is guaranteed even though many lanes run at once.

This is also what makes state safe under concurrency. A stateful operator — count, reduce, aggregate, a join, or a custom Processor reading and writing a store — only ever touches a given key from the single lane that owns it. Updates to one key are serialized on that lane; updates to different keys proceed in parallel on other lanes. No global lock, no contention on the common path. The reasoning is spelled out under State stores and durability on the architecture page; the State and thread safety concept page goes deeper on the rare cross-key case.

The flip side: parallelism is a function of how your keys spread across lanes. A stream where every record carries the same key runs effectively single-threaded — every record routes to one lane, by design, to preserve that key's order. Throughput on a stateful topology comes from having many distinct, evenly distributed keys, not from adding lanes alone.

Different keys run in parallel

Distinct keys are spread across lanes, so independent keys process concurrently. In a word-count topology, the, quick, and fox each own their running total on whichever lane their key maps to, and all three advance at the same time. The aggregation you write stays sequential per key — exactly what correctness needs — while the engine extracts parallelism across keys for free.

You do not wire any of this up. The DSL operators and your custom processors are written as if single-threaded per key; the engine provides the cross-key concurrency underneath. The grouping key you choose (groupBy, the join key, the record key on a source) is therefore also your unit of parallelism — pick a key with enough distinct values to keep lanes busy.

Lane count is decoupled from partition count

In standard Kafka Streams, processing parallelism is capped by the partition count of the input topics: one task per partition, with each stream thread running one or more tasks. To process more in parallel you add partitions — a topic-level change with downstream consequences.

StoatFlow breaks that coupling. The single consumer reads every partition of every source topic, and the engine then distributes work across however many lanes you configure. Lane count is independent of partition count — it scales with the cores on the machine, set once at startup.

Configure it with numLanes in StreamsConfig, or the stoatflow.lanes.count key in application.yaml. The default is max(2, available CPU cores).

val config =
    StreamsConfig(
        applicationId = "map-filter-example",
        bootstrapServers = "localhost:9092",
        defaultKeySerde = Serdes.String(),
        defaultValueSerde = Serdes.String(),
        numLanes = 6,
    )

var config =
    StreamsConfig.builder("map-filter-example", "localhost:9092")
        .defaultKeySerde(Serdes.String())
        .defaultValueSerde(Serdes.String())
        .numLanes(6)
        .build();

On the :runtime module the same setting lives in application.yaml:

stoatflow:
  application-id: map-filter-example
  bootstrap-servers: localhost:9092
  lanes:
    count: 6

How many lanes?

Lane count is a throughput-vs-overhead trade-off, not a correctness knob — ordering and exactly-once hold at any value. Some guidance at the architecture level:

• Start at the default (max(2, CPU cores)) and tune from measured throughput, not guesswork.
• More lanes than cores can help when lanes spend time blocked — on external calls, disk, or downstream systems — because parked lanes free their carrier thread for others. It rarely helps a purely CPU-bound topology, where you can't run more concurrent work than you have cores.
• More lanes cost more. Each lane carries its own queue and dispatch bookkeeping; past a point you pay overhead without gaining throughput.
• Your keyspace caps the benefit. Lanes beyond the number of distinct, well-distributed keys sit idle — they can't manufacture parallelism the data doesn't contain.

Pin numbers come from Benchmarks; the precise scheduling and dispatch behaviour is an implementation concern and stays in the source.

Virtual threads make blocking I/O cheap

Lanes run on virtual threads — JDK 21's GA primitive. A lane that blocks on a REST call, a database query, or an AI-inference response parks at near-zero cost while the JVM keeps making progress on every other lane. That changes how you write enrichment: a synchronous, blocking call inside a Processor is the idiomatic approach — no CompletableFuture chains, no reactive frameworks, no callback wiring just to keep throughput up while a call is in flight.

This processor enriches each flight by looking values up from a state store and forwarding the result — straight-line, blocking-style code. A remote lookup against an external service would read the same way; the lane simply parks until it returns.

(STATE_STORE_AIRPORT_INFO here is a global store, materialized by builder.globalTable(...), so the processor reads it without declaring it and gets it read-only. An ordinary store must be connected to the node — see Processor API.)

class AirportEnrichmentProcessor :
    ContextualProcessor<String, Flight, String, FlightEnriched>() {

    private lateinit var airportInfoStore: ReadOnlyKeyValueStore<String, AirportInfoI18n>

    override fun init(context: ProcessorContext<String, FlightEnriched>) {
        super.init(context)
        airportInfoStore = context.getStateStore(STATE_STORE_AIRPORT_INFO)
    }

    override fun process(record: Record<String, Flight>) {
        val flight = record.value ?: return
        // Blocking-style lookups; a remote enrichment call would read the same way.
        val depInfo = flight.departureAirport?.let { airportInfoStore.get(it) }
        val arrInfo = flight.arrivalAirport?.let { airportInfoStore.get(it) }
        context().forward(record.withValue(toFlightEnriched(flight, depInfo, arrInfo)))
    }
}

public class AirportEnrichmentProcessor
        extends ContextualProcessor<String, Flight, String, FlightEnriched> {

    private ReadOnlyKeyValueStore<String, AirportInfoI18n> airportInfoStore;

    @Override
    public void init(ProcessorContext<String, FlightEnriched> context) {
        super.init(context);
        this.airportInfoStore = context.getStateStore(STATE_STORE_AIRPORT_INFO);
    }

    @Override
    public void process(Record<String, Flight> record) {
        final Flight flight = record.value();
        if (flight == null) {
            return;
        }
        // Blocking-style lookups; a remote enrichment call would read the same way.
        final String dep = flight.getDepartureAirport();
        final String arr = flight.getArrivalAirport();
        final AirportInfoI18n depInfo = dep != null ? airportInfoStore.get(dep) : null;
        final AirportInfoI18n arrInfo = arr != null ? airportInfoStore.get(arr) : null;
        context().forward(record.withValue(toFlightEnriched(flight, depInfo, arrInfo)));
    }
}

See State and thread safety for the rules on what a lane may touch, and Processor API for the full processor surface.

In-memory re-keying instead of repartition topics

When the topology changes a record's key — selectKey, groupBy, or a key-changing join — that record may now belong to a different lane. StoatFlow re-hashes the new key and routes it to the lane that owns it in-memory, between lanes. There is no internal repartition topic, no extra serialization round-trip, and no broker hop.

In Kafka Streams the same key change forces a write to a repartition topic and a re-read on the other side. StoatFlow's single-instance model removes that round-trip entirely — re-keying is an in-process handoff. The output is identical; the path is shorter.

The e-commerce daily-summary topology re-keys twice — once on groupBy to aggregate per customer, then again coming out of the windowed table — and both are in-memory handoffs:

// Re-key the merged stream by customerId for aggregation.
val windowedStream =
    mergedStreams
        .groupBy(
            { _, v -> v.purchase?.customerId ?: v.webActivity.customerId },
            Grouped.with("grouped-by-customer-id", stringSerde, webActivityOrPurchaseSerde),
        )
        .windowedBy(dailyWindows)

// ...aggregate, then re-key back to customerId on the way out.
val dailyAggregatesStream =
    dailyAggregation
        .mapValues({ wk, v -> /* set date from window start */ }, Named.`as`("daily-aggregates-mapvalues"))
        .toStream({ wk, _ -> wk.key }, Named.`as`("daily-aggregates-to-stream"))

// Re-key the merged stream by customerId for aggregation.
TimeWindowedKStream<String, WebActivityOrPurchase> windowedStream = mergedStreams
    .groupBy(
        (k, v) -> v.purchase() != null ? v.purchase().customerId() : v.webActivity().customerId(),
        Grouped.with("grouped-by-customer-id", stringSerde, webActivityOrPurchaseSerde))
    .windowedBy(dailyWindows);

// ...aggregate, then re-key back to customerId on the way out.
KStream<String, DailyAggregates> dailyAggregatesStream = dailyAggregation
    .mapValues((wk, v) -> /* set date from window start */, Named.as("daily-aggregates-mapvalues"))
    .toStream((wk, v) -> wk.getKey(), Named.as("daily-aggregates-to-stream"));

After a re-key, the new key gets its own owning lane — the affinity property travels with the record across the handoff, so one key is still never processed by two lanes at once. That is what keeps state safe: aggregations, joins and suppression downstream of a groupBy still see a single writer per key.

Ordering is the narrower claim. Records that shared the old key stay on one lane throughout and keep their relative order. Records that a re-key newly brings together under one key arrived on different lanes, and those lanes run concurrently — so their relative order at the new key is not guaranteed, exactly as in Kafka Streams, where the same records would have been processed by different tasks. If you need a total order per key downstream of a merging re-key, you need it at the source: make the grouping key a function of the source key.

Exactly-once alignment does travel across the handoff: where the topology crosses a sub-topology boundary, a record that runs ahead of the commit barrier is briefly held on the receiving lane until the barrier reaches it there — so the committed cut stays exact across the boundary, not only within a single lane (see Exactly-once). This is also why your grouping key is your parallelism unit: the lane spread of a groupBy aggregation follows the distribution of the grouped key, not the source key.

Where the handoff actually happens

A re-key does not, by itself, trigger the handoff. StoatFlow inserts one where a downstream operator genuinely needs it — the same rule Kafka Streams uses to decide whether to materialise a repartition topic:

• grouped aggregations (groupBy(...).count/reduce/aggregate, windowed and session included),
• joins (stream–stream, stream–table, table–table, foreign-key),
• toTable() and any operator that materialises a per-key store,
• an explicit repartition(), which always forces one.

process() / processValues() are deliberately not on that list — Kafka Streams never repartitions before a Processor API node either, even when it has connected stores. StoatFlow did until 1.0.0; stoatflow.topology.processor-api-key-affinity: presumed restores it. The trade-off after a many-to-one re-key is real and is spelled out in Key affinity after a re-key — including the shape StoatFlow refuses to compile.

Everything else runs where it is: selectKey → mapValues → to() re-keys the record and writes it out on the source-key lane, with no handoff at all, because nothing downstream cares which lane it is on. That is one fewer set of lanes, queues and barrier stages than a boundary would cost.

The fix is the same idiom you would use in Kafka Streams: insert an explicit repartition() where you want the re-spread. stream.selectKey { ... }.repartition().mapValues { expensive(it) } forces the handoff and fans the work back out.

Genuinely null keys are unaffected — they are round-robined across lanes rather than hashed. An empty but non-null key is not: it hashes deterministically to one lane.

To restore the pre-1.0.0 behaviour globally (a handoff at every key change), set stoatflow.topology.sub-topology-split: eager. To restore it only at Processor API nodes, set stoatflow.topology.processor-api-key-affinity: presumed.

Where to go next

• Architecture — the single-instance engine, lane dispatch, and commit barriers in one place
• State and thread safety — what a lane may touch, and the cross-key exception
• Exactly-once — how per-lane processing commits atomically
• Processor API — writing custom processors that run on lanes
• Benchmarks — measured throughput across lane counts and workloads
• How StoatFlow differs from KS — the partition-coupling and repartition-topic contrasts in full