Your timer fired on another thread

StoatFlow promises that a key's timer callbacks run with the same serialised execution as process(...) for that key: one lane, one key's events at a time, so onTimer can read and write the key's state without a lock. The lane is chosen by hashing the key's bytes. A timer's lane is chosen the same way — by hashing the timer key's bytes, which the engine serialises again, with the processor's timer key serde. When those two byte strings differ, the timer lands on another lane and runs concurrently with the key's records.

Behind a Processor API source declared with only a key deserializer, that is what happened. The timer key serde fell back to default.key.serde; if the default wrote different bytes than the topic's producer, the key's timer fired on another lane, and nothing said so. Our reproduction put key k0's records on lane 0 and its timer on lane 6. This post explains where the two byte strings come from, why Kafka Streams never has to care, and what StoatFlow does now: declare the serde, check the lanes at build time, and stop the application on a proven split.

TL;DR

• What: a record's lane is the hash of its key bytes; a timer's lane is the hash of its key re-serialised by the timer key serde. Behind addSource(name, keyDeserializer, …), that serde was default.key.serde — different bytes, different lane, same key.
• Why it matters: onTimer and process for the same key ran at the same time on two virtual threads, racing on the key's state. No error, no metric.
• The fix: declare the key serde — Topology.addSource(name, Consumed.with(keySerde, valueSerde), topics). A build-time rule, timer-key-lane-affinity, warns where you have not.
• The guarantee: at runtime StoatFlow compares lanes, not bytes. A timer registered for the key being processed that picks another lane throws a fatal TimerLaneAffinityException by default; a sampled check covers the rest and counts stoatflow.timer.lane.mismatch.total.
• The catch: the sample can miss a mismatch that only some keys show, and the timer key serde is part of the stored timer format — changing it strands timers already persisted.
• Docs: Timer key serde · Lanes and parallelism.

In this post

• A lane is a key's serialisation point
• Where the two byte strings diverge
• Why Kafka Streams never meets this
• Declare the key serde
• Prove the split, then stop
• Catch it in a unit test
• Timers for another key
• Tradeoffs and limits

A lane is a key's serialisation point

A StoatFlow application is one process. Inside it, records are routed to lanes — StoatFlow's unit of parallel processing — by key: the engine hashes the key's bytes and takes the result modulo the lane count, so the same key always lands on the same lane. Each lane runs on its own virtual thread and processes its keys' records one at a time, in the order Kafka delivered them. That is what makes per-key state safe without locks: two records of one key are never processed at the same moment.

The bytes that are hashed depend on where the record is. In the first stage of a topology they are the topic's raw key bytes, exactly as the producer wrote them — no serde involved. Behind a re-key, where a record crosses to another lane in memory, they are the bytes the boundary's key serde writes.

A timer has its own bytes. When a processor registers or deletes a timer, the engine serialises the timer key with the processor's timer key serde: those bytes are what the timer store persists, what onTimer gets decoded from, and what picks the lane the timer fires on (behind a windowed aggregation, the inner key's bytes). Nothing on the timer path reads the bytes the record was laned by. So "a key's timers run on its records' lane" holds only if the timer key serde writes exactly the bytes the record was laned by.

Where the two byte strings diverge

The Kafka Streams-shaped Processor API declares a source with deserializers:

val topology = Topology()
    .addSource("events", keyDeserializer, eventDeserializer, "events")
    .addProcessor("sessionize", { SessionProcessor() }, "events")

A deserializer can read a key but not write one. With nothing else declared on the records' path, the timer key serde fell back to default.key.serde. Two outcomes followed. If the default could not serialise the key type at all, the first timer registration threw a ClassCastException — loud, and easy to fix. If it serialised the key successfully but differently from the producer, the timer was registered, persisted and fired, on whatever lane its bytes hashed to.

Different bytes are common. A producer writes the key as JSON — "k0", quotes included — and the source's deserializer strips the quotes; the default String serde then writes k0 without them. Other causes: a Schema Registry serde whose id has evolved, header-mode input, or a boundary laned by a different key serde than the one the timer uses.

We built the case on a real broker, red first: eight lanes, JSON-quoted keys on the wire, a deserializer that unquotes them, default.key.serde set to String. The test derives keys whose two encodings hash to different lanes rather than hard-coding them. Result: key k0's records ran on lane 0 while its timer fired on lane 6, with no log line and no metric. The control — keys whose encodings happen to share a lane — behaved.

The application kept running and broke its promise: onTimer for k0 ran on lane 6 while lane 0 processed k0's records. Any read-modify-write on the key's state could interleave: a session processor closing a session from onTimer while process extends it, a timeout firing while the record that should have cancelled it is half-processed. Global stores do not make that safe — the lane is the key's serialisation point, and the timer had left it.

Why Kafka Streams never meets this

Kafka Streams has no keyed timers — only punctuators, which belong to a task, not a key — and no in-memory boundaries: when a downstream operation needs a changed key's partitioning — a groupBy, a join, an explicit repartition() — the records go through a repartition topic. A source declares its key with a Deserializer, and a key is serialised only where it is written to a topic or a store. So Kafka Streams never has to re-serialise a key it only read, and never has to agree with itself about which bytes a key has.

StoatFlow added both features on purpose. Keyed timers are what make per-key timeouts and debounce windows natural, as in Flink. In-memory re-keying is what removes the repartition topics. Together they make the key's bytes decide which thread runs the code, as well as how the key is stored, and that needs a declared serde instead of a default.

Declare the key serde

The fix is to say what the key is. StoatFlow adds a Consumed-taking addSource:

import io.stoatflow.core.topology.Consumed
import io.stoatflow.core.topology.Topology
import org.apache.kafka.common.serialization.Serdes

val topology = Topology()
    .addSource("events", Consumed.with(Serdes.String(), eventSerde), "events")
    .addProcessor("sessionize", { SessionProcessor() }, "events")

The Consumed applies exactly as it does for StreamsBuilder.stream(...) — serdes, timestamp extractor or watermark strategy, offset reset, lane overrides — and a Pattern overload takes a topic pattern. addGlobalStore(...) and addReadOnlyStateStore(...) get the same Consumed twin, so a state-update processor's keys, timers included, use the source's own codec; their sources always read from the beginning, so an offset reset other than earliest is refused there. A source with a ByteArrayDeserializer key counts as declaring Serdes.ByteArray(), since the bytes it reads are the bytes it would write. The deserializer overloads are unchanged.

Behind the source, the timer key serde is the key serde declared for the key a processor receives: upstream on a Consumed, Grouped or Repartitioned, or downstream before any key change on a Repartitioned or Produced, otherwise default.key.serde. When nothing on the records' path declares the key, an upstream DSL store's key serde counts too — a Materialized.withKeySerde(…) on the aggregation before toStream(). A Processor API node in between counts as a key change, so the source's Consumed reaches a timer processor fed by that source directly; behind another Processor API node, declare the serde downstream or set default.key.serde.

At build time, the timer-key-lane-affinity validation rule reports a timer-aware processor whose timer bytes may differ from its records' lane bytes — behind a deserializer-only source, or across a boundary laned by a different key serde. It defaults to warn, because a configured default may be exactly right; if default.key.serde already writes the topic's key bytes, set the rule to off.

This is one case of a wider change in the same release: the DSL now carries only the serdes you declared. Defaults are applied once, from the application's StreamsConfig, when the topology is compiled. StreamsBuilder and TopologyConfig no longer carry default serdes of their own — Kafka Streams parity — and a windowed key keeps its serde through every operator downstream of the aggregation.

Prove the split, then stop

A build-time rule can only say "may differ". At runtime StoatFlow checks what happens, and it compares lanes, not bytes: two encodings that differ but hash to the same lane are no split, and are never reported. That matters because a byte comparison would flag every harmless difference — and a check that cries wolf gets switched off.

The guard runs on every timer registered or deleted for the key being processed, on that record's lane. If the timer key's bytes pick another lane than the record's, the split is proven: this key's timers would run beside its records. The default verdict is error. The registration does not happen; StoatFlow logs the split at ERROR and throws TimerLaneAffinityException, a fatal StreamsException that bypasses processing.exception.handler, so the epoch aborts and its offsets are not committed. Catching it does not hide it — a processor that catches it around the timer call, or an upstream processor that catches it around context.forward, sees it rethrown once that processor returns.

The sampled check covers what the guard cannot see: processors that only register timers from punctuators or onTimer, timers for other keys, and processors that never time the record's own key. Before process, it re-serialises the record key with the timer key serde and compares lanes — for the first 64 records per processor and lane, then one in 4,096. The default verdict is warn: it logs once per processor and counts every split it samples on stoatflow.timer.lane.mismatch.total, pre-registered at zero so an alert on its increase sees the first hit. It is not fatal by default because it judges more than proven splits, and because some wire encodings legitimately differ from a re-serialisation: an evolved Schema Registry id, JSON written by another producer.

Neither check runs on a sub-topology with a single lane, where no split is possible, or on records whose lane a custom StreamPartitioner on repartition() chose without hashing the key.

Catch it in a unit test

The TopologyTestDriver runs every piped record through the sampled check and 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 core count, so the same test gives the same answer on a laptop and in CI. To judge over your production lane count and fail the test on a systematic mismatch:

val props = Properties().apply {
    put("application.id", "sessionize-test")
    put("bootstrap.servers", "unused:9092")
    put("num.stream.threads", "8")                          // your production lane count
    put("timer.lane-affinity.sampled-check", "error")
}

TopologyTestDriver(topology, props).use { driver ->
    val input = driver.createInputTopic("events", Serdes.String(), eventSerde)
    input.pipeInput("k0", event)   // a lane split throws here, before deployment
}

The driver is single-threaded, so it cannot show the race itself. It does not need to: it computes the two lanes and fails on the mismatch that would cause it.

Timers for another key

A StoatFlow timer may be for any key the timer key serde can serialise, not only the record being processed — from process, onTimer or a punctuator. A timer for key y, armed while processing key x, fires on y's lane, serially with y's records. That goes beyond Flink, which scopes a timer to the current key.

The cost is ordering across lanes. Registering or deleting y's timer from x's lane is not ordered against what y's own lane does to y's timers at the same moment. The timer set stays consistent: a write from a later commit wins over one from an earlier commit, even when it was made earlier, the same way while the application runs and after a restart. But which commit each write lands in depends on timing, and a replay after a failure may place them differently. The same holds for moving a named timer from another lane. When the order matters, use the Flink-style pattern: forward a record re-keyed to y, and arm the timer while processing that record, so every key's timers have one writer.

Tradeoffs and limits

The sample can miss. After the first 64 records per processor and lane it looks at one record in 4,096. A mismatch that every key shows is caught in the first few records; one that only a few keys show — a rare schema version, one producer among many — can go unseen for a long time. The guard has no such gap for timers registered on the key being processed, but it judges only those. Declaring the key serde removes the question instead of sampling for it.

Lanes depend on the lane count. A key whose two encodings share a lane at eight lanes may not at sixteen. A check that passed before a lane-count change can fail after it. That is another reason to declare the serde rather than rely on a lucky hash.

The timer key serde is part of the timer store's format. Changing it between deployments — for example, adding addSource(..., Consumed...) to an application that ran on default.key.serde — changes the bytes of every future registration, while timers already stored keep the old bytes. Those fail to decode when they fire, are never deleted, or, worst, decode as a different key, so onTimer acts on the wrong key's state. Named timers are keyed by those bytes too: after a change, a registration by name no longer replaces a timer stored under the old bytes, so both fire. Settle the key serde before you rely on persisted timers, and treat a change as a state migration.

A custom StreamPartitioner is outside the contract. It picks a record's lane without hashing the key bytes, so neither check can judge it, and timers behind it are not guaranteed to share their records' lane.

A proven split now stops the application. An application that ran with split timers on an earlier release now stops at its first proven split, until the serde is declared or the guard is set to warn. We chose a loud failure over a silent race; the build-time rule is there to make that failure happen in a code review rather than in production.

Read on:

• Timer key serde — the full contract, the two checks and their settings.
• Lanes and parallelism — key affinity, virtual threads and in-memory re-keying.
• Configuration reference — timer-lane-affinity-* and topology.validation.*.
• Scheduled sources: a topology source that isn't a topic — the other kind of callback-driven input.
• Three gates, one transaction: error handling in StoatFlow — why a fatal exception aborts the epoch instead of reaching the handler.