The Processor API

Write custom record-by-record logic with Processor and FixedKeyProcessor — forward, access state, schedule punctuators, and register key-based timers.

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:

Interface	Attach with	Can change key?	Forward
`Processor<KIn, VIn, KOut, VOut>`	`KStream.process(...)`	Yes	new key + value
`FixedKeyProcessor<KIn, VIn, VOut>`	`KStream.processValues(...)`	No (key fixed)	value only

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.

process() and processValues() do not open a sub-topology boundary of their own, even after a re-key and even with connected stores — matching Kafka Streams exactly. Before 1.0.0 StoatFlow was stricter here and always opened one; that was the single remaining place a ported topology's describe() showed an extra sub-topology, and it is gone. Set topology.processor-api-key-affinity: presumed to restore the old boundary. What this costs you after a many-to-one re-key is below. See also Lanes and parallelism.

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.

One instance per lane — like a Kafka Streams task. A key-affinity lane is StoatFlow's analog of a KS task/stream-thread, and each gets its own single-threaded processor instance. So your instance fields are never raced — no synchronization needed, exactly as in KS. Three consequences: (1) a punctuator, which fires once over global state, cannot see another lane's instance fields; (2) after a many-to-one re-key, one key's records can arrive on several lanes, so per-key state in an instance field is spread across those lanes' instances — again exactly as in KS, where it is spread across tasks (see Key affinity after a re-key); (3) for state that must be shared across lanes or read by a punctuator, use a state store — StoatFlow's global-state channel — not a plain instance field. See State and thread safety and Attaching state stores.

Low-level Topology.addProcessor uses one extra instance. A processor added via the low-level Topology.addProcessor API gets a dedicated punctuator-owner instance in addition to the per-lane record instances — its init() is what registers context.schedule(...) punctuators, so a punctuator-only processor with no input records still fires. Expect init()/close() once per instance (twice at one lane), records only on lane instances, and punctuators only on the owner. The DSL process()/processValues() paths instead designate the first lane instance as owner (one instance at one lane). Deliberate divergence from KS's one-instance-per-task model — see the compatibility matrix.

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

The Named argument is optional but recommended — StoatFlow uses these stable names for the topology graph, metrics, and state-store identity. The same convention as the DSL operators in Your first app.

ProcessorContext

init(context) is called once before any records. 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:

Method	Behaviour
`forward(record)`	Forward a `Record<KOut, VOut>` — the record's timestamp and headers ride to the sink (set them via `record.withTimestamp(...)` / `record.withHeaders(...)`)
`forward(key, value)`	Forward using the current record's timestamp and headers
`forward(key, value, timestamp)`	Forward with an explicit timestamp
`forward(record, childName)` / `forward(key, value, childName)`	Forward to a specific named child node

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.

Record metadata and time

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

Method	Returns
`timestamp()`	Timestamp of the current record
`headers()`	The source record's headers (`org.apache.kafka.common.header.Headers`); `record.headers()` carries the same. Empty in timer and punctuator callbacks
`topic()` / `partition()` / `offset()`	Source coordinates of the current record
`applicationId()`	The configured application ID
`currentWatermarkMs()`	Global event-time progress (min watermark across non-idle partitions); may be `Long.MIN_VALUE` before any events
`currentStreamTimeMs()`	KS-compatible alias for `currentWatermarkMs()`
`currentSystemTimeMs()`	Wall-clock time (`System.currentTimeMillis()`)
`getStateStore(name)`	A state store by name

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.

State stores are epoch-scoped and accessed only from within process(...), onTimer(...), and punctuators — not from init background threads of your own making. Reads see your own uncommitted writes; the engine flushes and makes them durable at commit barriers. See State and thread safety.

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

You can only read a store you declared.getStateStore(name) throws StoreNotConnectedException — a fatal, non-recoverable error — if name is not connected to this processor node. The message names the processor, the store, and what the node actually declares. Two categories are exempt and need no declaration: global stores (addGlobalStore(...) and globalTable(...)), which are handed to an undeclaring processor read-only — declare one anyway and you get write access, a StoatFlow extension Kafka Streams rejects outright; and read-only stores (addReadOnlyStateStore(...)).

Note the failure surfaces on the first record rather than at startup, because StoatFlow creates one processor instance per lane on demand — including the init where the lookup usually lives. Kafka Streams reports the same mistake at task startup. It is deliberately not routable to processing.exception.handler: a CONTINUE or DLQ handler cannot swallow a topology-configuration error.

Shared or punctuator-visible state goes in a store, not an instance field. Because each lane has its own processor instance, state that must be visible across lanes — or read/written from a punctuator (punctuators fire once over global state) — belongs in a state store, StoatFlow's global-state channel. It's the same mechanism the built-in windowed operators use (lanes write in process; the punctuation lane reads to emit). Punctuators have full read/write store access. For ephemeral scratch that doesn't need to survive a restart, use an in-memory, logging-disabled store:

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

This is the drop-in replacement for a Kafka Streams processor that kept cross-record state in a plain instance field and read it from a punctuator.

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:

Shape	What happens
No re-key upstream	Unchanged — one lane per key.
Injective re-key (`k → "user-$k"`, 1:1)	Unchanged — distinct new keys still map to distinct lanes. Safe for free.
Node declares a timer (`declaredTimerTypes()` non-empty)	Unchanged — the boundary is kept, because the timer service hashes the new key while records lane by the old one.
Many-to-one re-key (several old keys → one new key)	Records for that new key arrive on several lanes and are processed concurrently.

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.

Neither compute() nor a lock closes this.compute(key, fn) and merge(...) are atomic within a commit epoch. Across a barrier rotation they are not, and no user-level lock can make them be: a lane still draining epoch N is structurally forbidden from reading a value another lane wrote into epoch N+1 — allowing it would import uncommitted state into a committed transaction. So both writes can derive from the same base and one is durably lost, with the lock held. A lock makes the safe ordering deterministic when it already exists; it cannot create it.

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:

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.
Set topology.processor-api-key-affinity: presumed to restore the boundary for every Processor API node in the application.
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. 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.

Eliding the boundary also puts the node in the upstream sub-topology, so it inherits that sub-topology's lane count — including a Consumed.withNumberOfLanes(...) set on the source. Under presumed it would have started its own. If you tuned lanes per sub-topology, re-check them after the flip.

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()).

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:

Constant	Alias	Fires when	Timestamp passed
`TimeNotion.STREAM_TIME`	`EVENT_TIME`	the watermark advances past the next interval	the current watermark
`TimeNotion.WALL_CLOCK_TIME`	`PROCESSING_TIME`	system time reaches the next interval	the current system time

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.

Punctuator mode

PunctuatorMode controls how a punctuator interacts with commit barriers:

Mode	Behaviour
`BLOCKING` (default)	The next commit waits for the punctuator to finish; records it forwards commit atomically with that barrier. Strong consistency, higher latency.
`NON_BLOCKING`	The punctuator runs asynchronously; records it forwards join whichever barrier is active when they're enqueued. Lower latency, no atomicity guarantee across the run. Best for fire-and-forget work (monitoring, metrics).

Punctuators execute on a dedicated punctuation lane per sub-topology (not on the per-key record lanes), with full state-store access in an epoch-aligned context. Because they run concurrently with the record lanes, a punctuator's read-modify-write on a key is not serialized against that key's record processing. For per-key mutations on a time trigger, prefer a timer (next section), whose onTimer callback runs in the same per-key serialized context as process(...). Each punctuator also has locked-run semantics: if a run is still in progress when the next interval fires, that trigger is skipped.

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();
        }
    }
}

context.schedule(...) also has an anchored overload that takes a startTime: Instant, snapping fire times to a fixed grid aligned to that anchor (KIP-1146) for deterministic, cron-like cadence. For topology-level periodic record generation that isn't tied to a processor, see Scheduled sources.

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. This follows Flink's timer semantics and is 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.

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

`TimerService<K>` method	Effect
`registerEventTimeTimer(key, timestamp)`	Fire when the watermark advances past `timestamp`
`registerProcessingTimeTimer(key, timestamp)`	Fire when wall-clock time reaches `timestamp`
`deleteEventTimeTimer(key, timestamp)`	Cancel a pending event-time timer
`deleteProcessingTimeTimer(key, timestamp)`	Cancel a pending processing-time timer
`currentWatermark()` / `currentProcessingTime()`	Read the current stream time / wall-clock time

Timers are deduplicated by (key, timestamp) — registering the same pair twice has no extra effect, and the timer fires exactly once.

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").delete(key)
        }
    }
}

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.delete(key);
        }
    }
}

Pick event-time timers for data-driven timeouts (sessionization, watermark-aligned flushing) — they advance with your data and are deterministic on replay. Pick processing-time timers for wall-clock deadlines that must fire regardless of data flow (the news-portal "publish at scheduled time" pattern). A persistent (RocksDB) timer backend survives restarts; with an in-memory backend, pending timers are rebuilt from your own state on recovery.

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.

Return ProcessorWrapper.asWrapped(supplier) / asWrappedFixedKey(supplier) to pass a node through unchanged. The default is NoOpProcessorWrapper, which does exactly that; with no wrapper configured the runtime record path is untouched, so there is no cost to leaving the key unset.

What a wrapper can and cannot do


Your Processor-API nodes (`process` / `processValues` / `addProcessor`)	Full control — decorate, replace the processor outright, or add state stores via the wrapped supplier's `stores()`
DSL operator nodes (`mapValues`, `filter`, aggregations, joins, windows)	The record path: observe, decorate, or short-circuit. A decorator may rewrite the record before delegating — `record.withTimestamp(...)` / `withHeaders(...)` flow through to the operator, downstream nodes, and the sink. Replacing a DSL operator with a different processor is rejected at build time — they are type-erased internally
`init` / watermark / suppress-flush callbacks on DSL nodes	Not routed through the wrapper (they aren't on the `Processor` surface) — in practice: window/session close emissions, the barrier suppress flush, and FK-join/suppress restore. A DSL-node decorator's context also cannot `schedule()` or use `timerService()`. Your timers are unaffected: keyed timers only exist on Processor-API nodes, which are wrapped in full
Global / read-only store update processors	Not wrapped — Kafka Streams doesn't wrap them either

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

Joins

Combine two streams or tables on a shared key — stream-stream windowed joins, stream-table enrichment lookups, table-table joins, foreign-key joins, and co-grouping.

Scheduled sources

Generate records on an interval or a cron schedule with StreamsBuilder.scheduled() — a topology source that emits without consuming from Kafka.