StoatFlow

Lanes and parallelism

How key-affinity lanes give StoatFlow per-key ordering and cross-key parallelism — decoupled from partition count, cheap to scale with cores, and friendly to in-line blocking I/O.

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,
    )

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:

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)))
    }
}
Virtual threads scale blocking I/O; they don't make CPU-bound work faster. A lane crunching numbers occupies a real carrier thread for the whole computation — concurrency there is still bounded by cores. Lanes win when work waits.

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

After each re-key, the new key's lane assignment again guarantees per-(new-)key ordering — the affinity property travels with the record across the handoff. 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 to go next

Exactly-once semantics

How StoatFlow's commit barrier delivers end-to-end exactly-once — one Kafka transaction commits state, output, and offsets atomically — plus crash recovery and the at-least-once trade-off.

State and thread-safety

How StoatFlow's global state model works — why concurrent access across lanes is safe, when you need the key-lock utility, and which store types are available.