The Processor API

When the declarative DSL operators don't cover what you need, drop down to the Processor API: a record-by-record interface where you control forwarding, state access, periodic work, and timers directly. It mirrors the Kafka Streams Processor API, so code written for KS ports with only an import change. For the engine model behind state access and timer execution, see Architecture and Lanes and parallelism.

Processor vs FixedKeyProcessor

Two interfaces, chosen by whether your logic changes the record key:

Use FixedKeyProcessor whenever you only transform values — a fixed key keeps the record on the lane that already owns it, so per-key state stays serialised without any re-keying at all. Reach for Processor when you need to emit records under a different key, and read Key affinity after a re-key before you keep per-key state in one.

Both interfaces are in io.stoatflow.core.processor. The convenience base classes ContextualProcessor and ContextualFixedKeyProcessor store the context for you so you don't have to keep a field; context() returns it after init.

A value-only processor

FixedKeyProcessor transforms the value and forwards it under the original key. process(record) runs once per input record; forward zero records to filter the record out, or more than one to fan out.

import io.stoatflow.core.processor.ContextualFixedKeyProcessor
import io.stoatflow.core.processor.FixedKeyRecord

class UppercaseProcessor : ContextualFixedKeyProcessor<String, String, String>() {
    override fun process(record: FixedKeyRecord<String, String>) {
        val value = record.value
        if (value.isNotBlank()) {
            // key and timestamp are preserved automatically
            context().forward(value.uppercase())
        }
    }
}

import io.stoatflow.core.processor.ContextualFixedKeyProcessor;
import io.stoatflow.core.processor.FixedKeyRecord;

public class UppercaseProcessor
        extends ContextualFixedKeyProcessor<String, String, String> {

    @Override
    public void process(FixedKeyRecord<String, String> record) {
        String value = record.value();
        if (value != null && !value.isBlank()) {
            // key and timestamp are preserved automatically
            context().forward(value.toUpperCase());
        }
    }
}

A key-changing processor

Processor can emit records under a different key. KStream.process(...) is a key-changing operation, so downstream stateful operators see the new key.

import io.stoatflow.core.processor.ContextualProcessor
import io.stoatflow.core.processor.Record

class RouteByTypeProcessor : ContextualProcessor<String, Event, String, Event>() {
    override fun process(record: Record<String, Event>) {
        val event = record.value
        // re-key on the event type, keeping the original timestamp
        context().forward(Record(event.type, event, record.timestamp))
    }
}

import io.stoatflow.core.processor.ContextualProcessor;
import io.stoatflow.core.processor.Record;

public class RouteByTypeProcessor
        extends ContextualProcessor<String, Event, String, Event> {

    @Override
    public void process(Record<String, Event> record) {
        Event event = record.value();
        // re-key on the event type, keeping the original timestamp
        context().forward(new Record<>(event.type(), event, record.timestamp()));
    }
}

Suppliers

process and processValues take a supplier, not a processor instance — the engine creates one instance per key-affinity lane, calling get() for each:

• ProcessorSupplier<KIn, VIn, KOut, VOut> — get() returns a Processor.
• FixedKeyProcessorSupplier<KIn, VIn, VOut> — get() returns a FixedKeyProcessor.

get() must return a fresh instance every time (Kafka Streams' contract; a supplier that hands back the same reference twice fails loudly). Both are functional interfaces, so a lambda or method reference (::new) works. They also expose a stores() method (default: empty) for declaring state stores inline with the processor — see Attaching state stores.

val routed: KStream<String, Event> =
    events.process({ RouteByTypeProcessor() }, Named.`as`("route-by-type"))

val shouted: KStream<String, String> =
    lines.processValues({ UppercaseProcessor() }, Named.`as`("uppercase"))

KStream<String, Event> routed =
    events.process(RouteByTypeProcessor::new, Named.as("route-by-type"));

KStream<String, String> shouted =
    lines.processValues(UppercaseProcessor::new, Named.as("uppercase"));

ProcessorContext

init(context) is called once per instance. As in Kafka Streams, the first instance is initialised when the application starts — before any record arrives, with or without input (in tests: when the TopologyTestDriver is built); each further lane's instance on that lane's first record. In HA active/standby mode init runs when a pod becomes active, never on a standby. Inside init, context.timestamp() is the current wall-clock time (the driver's clock in tests). Store the context (or extend a Contextual* base class and use context()). The context gives you forwarding, record metadata, time access, state stores, punctuator scheduling, and the timer service.

Forwarding

ProcessorContext (key-changing) forwards a full Record or a key/value pair:

FixedKeyProcessorContext forwards values only — the key is fixed: forward(value), forward(value, timestamp), and the FixedKeyRecord / childName variants. Call key() on the context to read the (immutable) key.

A child connected to the same parent twice (topology.addProcessor("x", supplier, "p", "p")) receives a plain forward once per connection and a forward(record, "x") once, as in Kafka Streams.

A node with several parents (addProcessor("x", supplier, "a", "b")) may receive different key types from each. When its parents sit in different sub-topologies, StoatFlow hands each record over to the node's own lanes by its key, so one key reaches one lane whichever parent sent it. The key is hashed with the key serde the parents declare; if they declare different ones, each parent's records are hashed with that parent's own serde, and the build logs a warning naming the node — the same key may then reach it on two lanes. Declare one key serde for every input of such a node when its parents carry the same keys.

Record metadata and time

Both contexts expose the metadata of the record currently being processed and the engine's time notions:

ProcessorContext additionally has isLate() — true when the current record's timestamp is below the watermark. It's only meaningful during process(...); in timer and punctuator callbacks it returns false. See Event time and watermarks.

Attaching state stores

A processor reads and writes a store by name via context.getStateStore(name). The store must be connected to that processor — asking for one that isn't throws StoreNotConnectedException, exactly as Kafka Streams does. There are two ways to connect one, and they union:

Option A — declare stores in the supplier

Override stores() on the supplier to return the StateStoreBuilders the processor needs. The engine creates and connects them automatically. Build store builders with the Stores factory in io.stoatflow.core.state.

import io.stoatflow.core.processor.ProcessorSupplier
import io.stoatflow.core.state.StateStoreBuilder
import io.stoatflow.core.state.Stores
import org.apache.kafka.common.serialization.Serdes

val supplier =
    object : ProcessorSupplier<String, Event, String, Event> {
        override fun get() = SessionProcessor("sessions")

        override fun stores(): Set<StateStoreBuilder<*>> =
            setOf(
                Stores.keyValueStoreBuilder(
                    Stores.persistentKeyValueStore("sessions"),
                    Serdes.String(),
                    sessionSerde,
                ),
            )
    }

val out = events.process(supplier, Named.`as`("sessionize"))

import io.stoatflow.core.processor.Processor;
import io.stoatflow.core.processor.ProcessorSupplier;
import io.stoatflow.core.state.StateStoreBuilder;
import io.stoatflow.core.state.Stores;
import org.apache.kafka.common.serialization.Serdes;

import java.util.Set;

ProcessorSupplier<String, Event, String, Event> supplier =
    new ProcessorSupplier<>() {
        @Override
        public Processor<String, Event, String, Event> get() {
            return new SessionProcessor("sessions");
        }

        @Override
        public Set<StateStoreBuilder<?>> stores() {
            return Set.of(
                Stores.keyValueStoreBuilder(
                    Stores.persistentKeyValueStore("sessions"),
                    Serdes.String(),
                    sessionSerde));
        }
    };

KStream<String, Event> out = events.process(supplier, Named.as("sessionize"));

Option B — register on the builder

Register the store on the StreamsBuilder with addStateStore(...), then name it in the process(...) call — registering alone does not connect it to any processor. persistentKeyValueStore is the recommended store type for production (RocksDB, changelog-backed); inMemoryKeyValueStore is available for non-durable state. See State stores.

import io.stoatflow.core.state.Stores
import org.apache.kafka.common.serialization.Serdes

builder.addStateStore(
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("sessions"),
        Serdes.String(),
        sessionSerde,
    ),
)

// The trailing store name is what connects "sessions" to this processor.
val out = events.process({ SessionProcessor("sessions") }, Named.`as`("sessionize"), "sessions")

import io.stoatflow.core.state.Stores;
import org.apache.kafka.common.serialization.Serdes;

builder.addStateStore(Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("sessions"),
    Serdes.String(),
    sessionSerde));

// The trailing store name is what connects "sessions" to this processor.
KStream<String, Event> out =
    events.process(() -> new SessionProcessor("sessions"), Named.as("sessionize"), "sessions");

A lookup in init — where it usually lives — fails at startup, as in Kafka Streams: start() throws (in tests, the TopologyTestDriver constructor). A lookup in process() fails on the record that makes it. Either way it is deliberately not routable to processing.exception.handler: a CONTINUE or DLQ handler cannot swallow a topology-configuration error.

builder.addStateStore(
    Stores
        .keyValueStoreBuilder(Stores.inMemoryKeyValueStore("buffer"), Serdes.String(), valueSerde)
        .withLoggingDisabled(), // ephemeral — no changelog topic
)

Sharing a store between processors

A processor connected to a store that another part of the topology writes — a table()'s store, say, or another processor's — is placed with that store's writer, as Kafka Streams puts processors that share a store into one sub-topology. With no re-key between them, a key reaches the same lane in both, so their writes to that key are serial. Only writers pull nodes together: a node that merely reads the store, such as a join, is not moved.

A re-key between two processors that connect the same store puts them in different sub-topologies. Their lanes cross commit barriers at different times, so a later write from one can land in an older commit than an earlier write from the other and be hidden by it — a lost update, under either processing guarantee. Kafka Streams forces such processors into one sub-topology (KAFKA-20464); StoatFlow places nodes by key, so it warns instead, with the presumed-cross-sub-topology-write rule (default warn, see the configuration reference). If only one of them writes the store and the others only read it, nothing is lost: set the rule to off. Otherwise connect the store from one sub-topology only, or let one processor own it and forward what the others need. The same shape among DSL writers is refused at build time.

Reading a store from another lane

Stores.lruMap(...) and Stores.inMemoryVersionedKeyValueStore(...) write directly instead of batching writes per commit, so their reads are always live. Under exactly_once a read of one from another lane can see a write that is not committed yet. When a Processor API node may read such a store across lanes — from a punctuator, or a node downstream of one in the same sub-topology, while record lanes write it; or from another sub-topology than the store's writer — the build warns with the presumed-cross-lane-read rule (default warn). A global or read-only store of either kind counts as read by every lane, because any processor may read one without declaring it.

To fix it, materialise the store with a default store supplier or, for versioned state, Stores.persistentVersionedKeyValueStore(...) — both are read as of the lane's commit — or run at_least_once. If the processor never reads the store that way, set the rule to off. The same read by a DSL join is refused at build time (joins under exactly-once), as is a DSL node that reads such a store on the punctuation lane — downstream of a LEFT or OUTER stream–stream join, say — while record lanes write it. A store that only the punctuation lane writes and reads, such as the table an OnWindowClose aggregation feeds, is fine.

Reading and writing the store

Look up the store in init, then use it in process (and onTimer). getStateStore is generic over the store type; in Kotlin pass the type argument, in Java assign to a typed variable. The same connection rule applies in every callback — process, onTimer and punctuators all see the node's declared set.

import io.stoatflow.core.processor.ContextualProcessor
import io.stoatflow.core.processor.ProcessorContext
import io.stoatflow.core.processor.Record
import io.stoatflow.core.state.KeyValueStore

class CountingProcessor : ContextualProcessor<String, String, String, Long>() {
    private lateinit var counts: KeyValueStore<String, Long>

    override fun init(context: ProcessorContext<String, Long>) {
        super.init(context)
        counts = context.getStateStore("counts")
    }

    override fun process(record: Record<String, String>) {
        val next = (counts.get(record.key) ?: 0L) + 1
        counts.put(record.key, next)
        context().forward(record.key, next)
    }
}

import io.stoatflow.core.processor.ContextualProcessor;
import io.stoatflow.core.processor.ProcessorContext;
import io.stoatflow.core.processor.Record;
import io.stoatflow.core.state.KeyValueStore;

public class CountingProcessor
        extends ContextualProcessor<String, String, String, Long> {

    private KeyValueStore<String, Long> counts;

    @Override
    public void init(ProcessorContext<String, Long> context) {
        super.init(context);
        this.counts = context.getStateStore("counts");
    }

    @Override
    public void process(Record<String, String> record) {
        long current = counts.get(record.key()) == null ? 0L : counts.get(record.key());
        long next = current + 1;
        counts.put(record.key(), next);
        context().forward(record.key(), next);
    }
}

Key affinity after a re-key

Normally a key's records all run on one lane, so per-key read-modify-write is serial and you need no locking. A many-to-one re-key immediately upstream of a Processor API node breaks that — and it is the one shape StoatFlow can detect at build time. (The other way to break it is to write a store under a key that is not the record key; see state and thread safety. That one is computed inside your processor at runtime, so no compiler can see it.)

Since 1.0.0 a Processor API node does not, by itself, require key affinity (topology.processor-api-key-affinity: off) — matching Kafka Streams, which never materialises a repartition before process() / processValues() even with connected stores. StoatFlow lanes records by the key they arrived with, so:

In the last row, per-key state is fragmented across lanes: instance fields stay race-free (one instance per lane), but a given key's state now lives in several of them — which is exactly what happens across tasks in Kafka Streams after the same re-key without a repartition(). State in a store is worse than fragmented, and it is worth being precise about why.

StoatFlow refuses to compile the provable case

When a re-key feeds a store-connected Processor API node with no boundary between them, the topology fails to build — in your own test suite, with no broker, because the check runs at topology-compile time:

Topology validation failed: 1 violation(s) configured as 'error'.

  - [papi-key-affinity-presumed] Processor API node 'papi' connects 'counts' and follows the
    re-key at 'rekey' with no sub-topology boundary between them, so records reach it on the
    lane of their OLD key. ...

Three remedies, in order of preference:

1. Insert repartition() before the node. In StoatFlow this is an in-memory lane hand-off, not a Kafka topic — no broker round-trip, no extra latency budget, no topic to provision. If you are arriving from Kafka Streams, this is much cheaper than it looks there.
2. Set topology.processor-api-key-affinity: presumed to restore the boundary for every Processor API node in the application.
3. Stop keying the store by the post-re-key key — keep it keyed by something that is still lane-stable.

If the re-key really is injective, downgrade the rule with topology.validation.papi-key-affinity-presumed: warn. Read-only stores (KIP-813) need no action — they cannot be written, so they are excluded from the check.

When the re-key is only presumed

process() and Topology.addProcessor() declare a key change unconditionally, because StoatFlow cannot see inside your Processor. So in a chain like process(parse).process(count, "store") — the shape a KSML-generated topology is made of — the "re-key" may not exist at all, and Kafka Streams compiles and runs it without a repartition.

Failing that build would be a refusal on evidence that does not exist, so it gets its own rule, papi-key-affinity-presumed-chain, defaulting to warn:

Topology validation [papi-key-affinity-presumed-chain]: Processor API node 'count' connects
'store' and follows the Processor API node 'parse' with no sub-topology boundary between them.
StoatFlow cannot see whether 'parse' changes the key ... but IF it does, and many-to-one, then ...

Only you can answer it. If parse does not change the key, nothing is wrong — set the rule to off. If it does, the remedies above apply: one key's records can then reach several lanes of one sub-topology, which cross commit barriers at different times, so two writes of that key can land in the wrong commit order and one is lost. Commit ordering protects a key only while it stays on one lane, so this warning is the only guard for that shape. Making the re-key explicit upstream (a selectKey) also moves the finding to the error-default rule, where it belongs.

What the build-time check cannot see

Two things.

Per-lane instance fields. No declaration can express them, so nothing can prove the shape. After a many-to-one re-key with the setting off, per-key state held in a processor's own fields is fragmented across lanes — exactly as it is across tasks in Kafka Streams. State held in a store is the case StoatFlow refuses to build, because that is the only place StoatFlow would be worse than Kafka Streams (KS fragments the value across N stores; StoatFlow can lose an update) rather than merely equal to it.

A store key that is not the record key. store.put(record.value().customerId, …) has the same cross-barrier window with no re-key anywhere, because the key you store under is not the key you are laned by. It is computed at runtime, so the compiler never sees it. State and thread safety covers what a lock does and does not buy you there.

Boundaries elided under off are logged once at startup, at INFO, naming the nodes — so an elided boundary is visible somewhere other than this page.

Punctuators

A punctuator is a callback that runs periodically — useful for heartbeats, metrics, or flushing buffered state. Schedule one in init via context.schedule(...). It returns a Cancellable you can use to stop it (for example in close()).

Because init runs when the application starts, a punctuator scheduled there starts with the application — a catch-up, heartbeat or TTL-cleanup punctuator runs after a restart even if no record arrives, as in Kafka Streams. When it fires is Kafka Streams' rule too — see Fire times.

schedule takes an interval, a TimeNotion, an optional PunctuatorMode, and the callback. The callback receives the trigger timestamp.

Time notion

TimeNotion is the same type used for timers and StreamsBuilder.scheduled(...). It carries both StoatFlow/Kafka-Streams and Flink naming for the same two concepts:

From Java, the PunctuationType object exposes STREAM_TIME / WALL_CLOCK_TIME and TimeDomain exposes EVENT_TIME / PROCESSING_TIME — both resolve to the same TimeNotion constants.

Fire times

Fire times follow Kafka Streams exactly:

• A STREAM_TIME punctuator is aligned to 0: it fires on the first data — the first watermark at or past 0 — and then at every multiple of the interval.
• A WALL_CLOCK_TIME punctuator first fires one interval after it was scheduled, then every interval on that grid.
• An anchored punctuator (schedule(interval, startTime, …), KIP-1146) first fires at startTime — or, if startTime is already behind the current time, at the first grid point after it.
• When time jumps past several grid points, the punctuator fires once, with the current time, and the next fire is the next grid point. Fire times never drift from the grid.

Inside the callback the context describes the punctuation, not the last record: context.timestamp() is the punctuation time (the value punctuate receives), context.headers() is empty, and context.topic() / partition() / offset() return "" / -1 / -1. A record built with context.timestamp() in a punctuator therefore carries the punctuation time — as in Kafka Streams.

Punctuator mode

PunctuatorMode controls how a punctuator interacts with commit barriers:

Example

import io.stoatflow.core.processor.Cancellable
import io.stoatflow.core.processor.ContextualProcessor
import io.stoatflow.core.processor.ProcessorContext
import io.stoatflow.core.processor.PunctuatorMode
import io.stoatflow.core.processor.Record
import io.stoatflow.core.processor.TimeNotion
import java.time.Duration

class HeartbeatProcessor : ContextualProcessor<String, Int, String, Int>() {
    private var schedule: Cancellable? = null

    override fun init(context: ProcessorContext<String, Int>) {
        super.init(context)
        // wall-clock heartbeat every 10 seconds
        schedule =
            context.schedule(Duration.ofSeconds(10), TimeNotion.WALL_CLOCK_TIME) { ts ->
                context.forward("heartbeat", ts.toInt())
            }
    }

    override fun process(record: Record<String, Int>) {
        context().forward(record.key, record.value)
    }

    override fun close() {
        schedule?.cancel()
    }
}

import io.stoatflow.core.processor.Cancellable;
import io.stoatflow.core.processor.ContextualProcessor;
import io.stoatflow.core.processor.ProcessorContext;
import io.stoatflow.core.processor.PunctuationType;
import io.stoatflow.core.processor.Record;
import java.time.Duration;

public class HeartbeatProcessor
        extends ContextualProcessor<String, Integer, String, Integer> {

    private Cancellable schedule;

    @Override
    public void init(ProcessorContext<String, Integer> context) {
        super.init(context);
        // wall-clock heartbeat every 10 seconds
        this.schedule = context.schedule(
            Duration.ofSeconds(10),
            PunctuationType.WALL_CLOCK_TIME,
            ts -> context.forward("heartbeat", (int) ts));
    }

    @Override
    public void process(Record<String, Integer> record) {
        context().forward(record.key(), record.value());
    }

    @Override
    public void close() {
        if (schedule != null) {
            schedule.cancel();
        }
    }
}

Timers

A timer fires once for a specific key at a specific instant, in event time or processing time. Unlike punctuators, timer callbacks run in the same per-key context as records — so onTimer(...) has full read/write state access and forwarding, with the same serialized execution as process(...) for that key. Like Flink's keyed timers — though a StoatFlow timer may be for any key, not only the current one — they are the right tool for per-key timeouts, debounce windows, and scheduled emissions.

Declaring timer types

A processor that uses timers must override declaredTimerTypes() to return the TimeNotions it will register. The engine uses this both to validate at runtime (registering an undeclared type throws IllegalStateException) and to optimise the hot path (it can skip watermark tracking when no processor declares event-time timers). The default is the empty set.

Timer key serde

Timer keys are serialized when a timer is registered, deleted and persisted, and deserialized when it fires. The timer's lane is chosen by hashing those bytes, the same way a record's lane is chosen by hashing its key bytes. A processor's timer key serde is the key serde declared for the key it receives — upstream (a Consumed, Grouped or Repartitioned serde) or downstream before any key change (a Repartitioned or Produced serde) — otherwise default.key.serde. When nothing on the records' path declares it, an upstream DSL store's key serde counts too: behind groupBy(…).count(Materialized.withKeySerde(…)).toStream(), the timer key serde is the Materialized one, the serde its records are laned by. A Processor API store never counts — its keys are whatever your processor writes. Where several branches reach the processor (a merge, or a Processor API node with several parents), a branch that changes the key declares nothing, and branches that declare different serdes leave the choice to default.key.serde. A Processor API node in between (process() / addProcessor()) counts as a key change, so a timer behind one resolves from downstream declarations or the default. Behind a windowed aggregation the timer key is a Windowed<K>, while the records keep the lane of their inner key K (a windowed aggregation is not a key change). Such a timer is laned by the inner key's bytes — through every windowing, for a key windowed more than once — and fires on that lane, so all of a key's windows, records and timers alike, stay on one lane.

For a Processor API source, declare the key serde with 🆕 Topology.addSource(name, Consumed.with(keySerde, valueSerde), topics). The Consumed applies exactly as it does for StreamsBuilder.stream(...): serdes, timestamp extractor or watermark strategy, offset reset and lane overrides. A Consumed name must equal the node name. A Pattern overload takes a topic pattern instead of topics.

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

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

Topology topology = new Topology()
    .addSource("events", Consumed.with(Serdes.String(), eventSerde), "events")
    .addProcessor("sessionize", SessionProcessor::new, "events");

addGlobalStore(...) and addReadOnlyStateStore(...) have the same Consumed-taking twin (storeBuilder, sourceName, consumed, topic, processorName, supplier), so a state-update processor's keys, timers included, use the source's own codec. Those sources always read from the beginning, so a Consumed offset reset other than earliest is refused. All three are StoatFlow extensions; the Kafka Streams-shaped (de)serializer overloads are unchanged.

addSource(name, keyDeserializer, valueDeserializer, topics) declares a deserializer, not a serde. A deserializer can read a key but not write one, so behind it the timer key serde falls back to default.key.serde, which must then serialize the key type to exactly the topic's key bytes. A default that cannot serialize the key type fails at the first timer registration with a ClassCastException naming the processor and the fix. A SerializationException from the serializer is rethrown unchanged, since it may be a transient Schema Registry failure.

The contract. A key's timers run on the same lane as its records, serially with them, whenever the timer-key-lane-affinity validation rule is silent. The one exception is a custom StreamPartitioner on repartition(), which picks a record's lane without hashing the key bytes.

• At build time, the timer-key-lane-affinity rule (default warn) reports a timer-aware processor whose timer bytes may differ from its records' lane bytes: one behind a source that declares only a key deserializer, or one entered across a sub-topology boundary laned by a different key serde. The fix is to declare the key serde — Consumed.with(Serdes.ByteArray(), …) for a byte-array source, or for a boundary the same key serde on the operator that opens it (Repartitioned / Grouped). If default.key.serde already writes exactly the topic's key bytes, set the rule to off.
• At runtime, StoatFlow serializes the timer key and checks which lane those bytes pick against the lane the record was laned to. Bytes that differ but land on the same lane are no split, so they are never reported. There are two checks:
  • The guard runs on every timer registered or deleted for the key being processed. A timer key that picks another lane — a default that encodes the key differently, header-mode input, Schema Registry id drift — would split that key's timers from its records. Its verdict is set by timer.lane-affinity.guard (YAML stoatflow.processing.timer-lane-affinity-guard): error (default) logs the split at ERROR and throws the fatal TimerLaneAffinityException (io.stoatflow.core.exception, a StreamsException), which bypasses processing.exception.handler, so the epoch aborts and its offsets are not committed. The timer is not registered, and catching the exception does not hide it: whether a process() catches it around the timer call or an upstream processor catches it around context.forward, StoatFlow rethrows it once that processor returns, and again when the record's traversal ends — the same holds for every failure that must stop the record, such as a FAIL verdict or a store-connection violation; warn logs once per processor and lets the timer register; off does not check. Every split counts on stoatflow.timer.lane.mismatch.total{processor}.
  • The sampled check re-serializes the record key before process (the first 64 records per processor and lane, then 1 in 4,096). Its verdict is set by timer.lane-affinity.sampled-check (YAML stoatflow.processing.timer-lane-affinity-sampled-check): warn (default) logs once per processor, counts every split on stoatflow.timer.lane.mismatch.total and keeps running; error throws TimerLaneAffinityException; off does not sample. It defaults to warn because it also judges processors that never register a timer for the record's own key, and some wire encodings legitimately differ from a re-serialization (an evolved Schema Registry id, JSON written by another producer).
  
  Neither check runs on sub-topologies with a single lane (no split is possible there), on records whose lane a repartition() StreamPartitioner picked, or for a registration from onTimer or a punctuator — only the record being processed is judged. Lanes depend on the sub-topology's lane count, so a key that shares a lane at one lane count may not at another; declaring the key serde removes the question.
• In tests, the TopologyTestDriver samples every piped record 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 cores — following the same two settings, and it applies a repartition() partitioner per record as the engine does. Set num.stream.threads to your production value to judge over exactly that, and timer.lane-affinity.sampled-check=error to make a systematic mismatch fail the unit test before deployment.

Registering and handling timers

Get the timer service in init via context.timerService(). Register timers in process (or onTimer), and handle them by overriding onTimer(timestamp, key, context).

Timers are deduplicated by (key, timestamp) — registering the same pair twice keeps one timer, and it fires once. That holds even when the timer is already due and its fire is queued on its lane, behind records still to be processed: a registration of the same (key, timestamp) from that lane that reaches it before it fires does not make it fire again. Deleting a timer from its lane cancels it, also when its fire is already queued: onTimer does not run. A registration or delete from another lane, or from a punctuator, follows the cross-lane rule below. In a sub-topology behind a re-key, a due timer's onTimer waits until the latest commit boundary has reached its lane, as re-keyed records do: it never runs in an older commit than the writes made before it became due.

A timer can be for any key, not only the record being processed — any key the timer key serde can serialize, from process, onTimer or a punctuator (including init). A timer for key y, armed while processing key x, fires on y's lane, serially with y's records. This goes beyond Flink, which scopes a timer to the current key, and Kafka Streams, which has no keyed timers. Order across lanes is not guaranteed: registering or deleting y's timer from x's lane or a punctuator is not ordered against what y's own lane does to y's timers at the same moment. The timer set stays consistent, but which write wins depends on timing: a write from a later commit wins over one from an earlier commit, even when the earlier one was made later — the same way while the application runs and after a restart. So a re-registration from another lane that belongs to a later commit than a fire already waiting, or running, on y's lane fires the timer once more, after it; one from the same commit or an earlier one does not. A delete from another lane follows the same order: from the same commit as a waiting fire or an earlier one it cancels the fire, but from a later commit it does not — the fire runs in its own commit, and the delete comes after it. This holds on the persistent timer backends (TwoTier, the default, and RocksDB); the Heap backend has no commits to order by and applies a write from another lane when it happens — a re-registration that meets a waiting fire is taken into it (one fire), and a delete cancels it. A replay after a failure may place the writes in different commits: a re-registration from another lane may then fire the timer once more or collapse into one fire, and a delete from another lane may cancel a fire it came after the first time, or the reverse. When order matters, use the Flink-style pattern: forward a record re-keyed to y and arm the timer while processing it, so each key has one writer.

A timer needs a key: every register* and delete* call with a null key throws IllegalArgumentException. As in Flink, a keyed timer belongs to a key, and records with a null key are spread across lanes rather than tied to one.

A timer can be re-registered from its own onTimer — through context.timerService() — in event time and in processing time alike, for a timeout that keeps rolling forward. Re-arming the very same (key, timestamp) from its onTimer creates a new timer that fires once more — at the next watermark advance or wall-clock tick; it is not deduplicated against the fire that armed it.

If onTimer throws, or the timer's key cannot be deserialized when it fires, the failure goes to the processing exception handler with a synthetic __timer source topic. The handler gets the timer's key, or the raw key bytes when the key itself could not be decoded. On CONTINUE, its dead letters are written and the timer is dropped. On FAIL — or if the handler itself throws, or its dead letters cannot be sent — the instance stops and the timer is kept, so with a persistent timer backend (TwoTier, the default, or RocksDB) it fires again after the restart; a Heap timer is lost on restart either way. A failing punctuator is routed the same way, with a synthetic __punctuator source topic.

The onTimer callback receives a TimerContext, which provides forward(...), getStateStore(name) (read/write), timerService() to register follow-up timers, the scheduled timestamp(), and timeDomain() to distinguish event-time from processing-time timers.

import io.stoatflow.core.processor.ContextualProcessor
import io.stoatflow.core.processor.ProcessorContext
import io.stoatflow.core.processor.Record
import io.stoatflow.core.processor.TimeNotion
import io.stoatflow.core.processor.TimerContext
import io.stoatflow.core.processor.TimerService
import io.stoatflow.core.state.KeyValueStore

private const val SESSION_TIMEOUT_MS = 30_000L

class SessionProcessor :
    ContextualProcessor<String, Event, String, Session>() {

    private lateinit var sessions: KeyValueStore<String, Session>
    private lateinit var timers: TimerService<String>

    override fun declaredTimerTypes(): Set<TimeNotion> = setOf(TimeNotion.EVENT_TIME)

    override fun init(context: ProcessorContext<String, Session>) {
        super.init(context)
        sessions = context.getStateStore("sessions")
        timers = context.timerService()
    }

    override fun process(record: Record<String, Event>) {
        val session = sessions.get(record.key)?.add(record.value)
            ?: Session.start(record.value)
        sessions.put(record.key, session)
        // (re)arm the session-timeout timer
        timers.registerEventTimeTimer(record.key, record.timestamp + SESSION_TIMEOUT_MS)
    }

    override fun onTimer(
        timestamp: Long,
        key: String,
        context: TimerContext<String, Session>,
    ) {
        val session = context.getStateStore<KeyValueStore<String, Session>>("sessions").get(key)
        if (session != null && session.lastActivity + SESSION_TIMEOUT_MS <= timestamp) {
            context.forward(key, session.complete())
            context.getStateStore<KeyValueStore<String, Session>>("sessions").put(key, null) // tombstone; we already read the value above
        }
    }
}

import io.stoatflow.core.processor.ContextualProcessor;
import io.stoatflow.core.processor.ProcessorContext;
import io.stoatflow.core.processor.Record;
import io.stoatflow.core.processor.TimeNotion;
import io.stoatflow.core.processor.TimerContext;
import io.stoatflow.core.processor.TimerService;
import io.stoatflow.core.state.KeyValueStore;
import org.jspecify.annotations.NonNull;

import java.util.Set;

public class SessionProcessor
        extends ContextualProcessor<String, Event, String, Session> {

    private static final long SESSION_TIMEOUT_MS = 30_000L;

    private KeyValueStore<String, Session> sessions;
    private TimerService<String> timers;

    @Override
    public @NonNull Set<TimeNotion> declaredTimerTypes() {
        return Set.of(TimeNotion.EVENT_TIME);
    }

    @Override
    public void init(ProcessorContext<String, Session> context) {
        super.init(context);
        this.sessions = context.getStateStore("sessions");
        this.timers = context.timerService();
    }

    @Override
    public void process(Record<String, Event> record) {
        Session existing = sessions.get(record.key());
        Session session = existing == null
            ? Session.start(record.value())
            : existing.add(record.value());
        sessions.put(record.key(), session);
        // (re)arm the session-timeout timer
        timers.registerEventTimeTimer(record.key(), record.timestamp() + SESSION_TIMEOUT_MS);
    }

    @Override
    public void onTimer(long timestamp, String key, @NonNull TimerContext<String, Session> context) {
        KeyValueStore<String, Session> store = context.getStateStore("sessions");
        Session session = store.get(key);
        if (session != null && session.lastActivity() + SESSION_TIMEOUT_MS <= timestamp) {
            context.forward(key, session.complete());
            store.put(key, null); // tombstone; we already read the value above
        }
    }
}

A complete processor app

Putting it together: register the store, attach the processor with process(...), and run it on the runtime. The processor reads its store by name, forwards results, and the runtime wires up Kafka, the HTTP/metrics server, and graceful shutdown — see Your first app for the runtime entry point.

import io.stoatflow.core.state.Stores
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 io.stoatflow.runtime.StoatFlowRuntime
import org.apache.kafka.common.serialization.Serdes

fun main() {
    val runtime = StoatFlowRuntime.fromConfig(
        topologyBuilder = { buildTopology(it) },
        configure = {
            streamsConfigOverrides {
                defaultKeySerde(Serdes.String())
                defaultValueSerde(Serdes.String())
            }
        },
    )
    runtime.start()
    runtime.awaitTermination()
}

private fun buildTopology(builder: StreamsBuilder) {
    builder.addStateStore(
        Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("counts"),
            Serdes.String(),
            Serdes.Long(),
        ),
    )

    builder
        .stream<String, String>("input", Consumed.`as`("source"))
        .process({ CountingProcessor() }, Named.`as`("count"), "counts")
        .to("counts-out", Produced.`as`<String, Long>("sink").withValueSerde(Serdes.Long()))
}

import io.stoatflow.core.state.Stores;
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;
import io.stoatflow.runtime.StoatFlowRuntime;
import org.apache.kafka.common.serialization.Serdes;

public class Main {

    public static void main(String[] args) {
        var runtime = StoatFlowRuntime.fromConfig(
            Main::buildTopology,
            builder -> builder.streamsConfigOverrides(cfg -> {
                cfg.defaultKeySerde(Serdes.String());
                cfg.defaultValueSerde(Serdes.String());
            })
        );
        runtime.start();
        runtime.awaitTermination();
    }

    private static void buildTopology(StreamsBuilder builder) {
        builder.addStateStore(Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("counts"),
            Serdes.String(),
            Serdes.Long()));

        KStream<String, String> input = builder.stream("input", Consumed.as("source"));
        input
            .process(CountingProcessor::new, Named.as("count"), "counts")
            .to("counts-out", Produced.<String, Long>as("sink").withValueSerde(Serdes.Long()));
    }
}

Processor wrapping (KIP-1112)

Sometimes you want one piece of behaviour applied to every processor in a topology — structured logging, per-node metrics, a tracing span, uniform error handling, or swapping in a mock while testing. Rather than editing every operator, configure a single ProcessorWrapper and StoatFlow applies it to every node — DSL operators and Processor-API nodes alike — when the topology is compiled.

public final class LoggingWrapper implements ProcessorWrapper {

    @Override
    public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wrapProcessorSupplier(
            String processorName, ProcessorSupplier<KIn, VIn, KOut, VOut> supplier) {
        return ProcessorWrapper.asWrapped(() -> {
            Processor<KIn, VIn, KOut, VOut> inner = supplier.get();
            return new Processor<>() {
                @Override public void init(ProcessorContext<KOut, VOut> ctx) { inner.init(ctx); }
                @Override public void process(Record<KIn, VIn> record) {
                    log.debug("{} <- {}", processorName, record.key());
                    inner.process(record);
                }
                @Override public void close() { inner.close(); }
            };
        });
    }

    @Override
    public <KIn, VIn, VOut> WrappedFixedKeyProcessorSupplier<KIn, VIn, VOut> wrapFixedKeyProcessorSupplier(
            String processorName, FixedKeyProcessorSupplier<KIn, VIn, VOut> supplier) {
        return ProcessorWrapper.asWrappedFixedKey(supplier);   // pass through
    }
}

Point processor.wrapper.class at it:

processor.wrapper.class=com.acme.LoggingWrapper

The Kafka Streams route — passing a TopologyConfig to the builder — works and is the portable choice:

StreamsConfig config = StreamsConfig.fromProperties(props).build();
StreamsBuilder builder = new StreamsBuilder(new TopologyConfig(config));

Unlike Kafka Streams, StoatFlow also honours a wrapper set on the runtime config alone — it compiles the topology inside start(), so KS's "too late, silently ignored" case doesn't exist here. If both are set to different classes, the TopologyConfig one wins and a warning names the shadowed value.

If your wrapper needs settings, implement configure(Map) — it is called once at instantiation with the config map, exactly as in Kafka Streams.

What a wrapper can and cannot do

Two behavioural notes worth knowing before you write a stateful wrapper:

• get() runs once per lane, not once per task. Each decorator instance is pinned to one lane and single- threaded, so per-instance fields are race-free — but they only see that lane's records. Put cross-lane counters on the wrapper object itself (and make it thread-safe), or push to an external sink.
• Node names are StoatFlow's — mapValues-0, process-0, or whatever you passed to Named.as(...) — not Kafka Streams' KSTREAM-MAPVALUES-0000000001. A ported wrapper that switches on node-name patterns compiles cleanly and then matches nothing.

Next steps

• State stores — store types, serdes, changelog backing, and querying.
• Scheduled sources — topology-level periodic record generation (StreamsBuilder.scheduled()).
• Testing — drive processors, advance time, and trigger punctuators with the in-memory test driver.
• State and thread safety — the execution model behind per-key serialized processing and timer callbacks.
• How StoatFlow differs from Kafka Streams — Processor API parity and the StoatFlow-specific extensions (timers, time notions).