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 — keeping the key fixed avoids a re-keying boundary in the topology. Reach for Processor when you need to emit records under a different key.

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 fresh instances as needed:

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

Both are functional interfaces, so a lambda or method reference 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"));

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>`
`forward(key, value)`	Forward using the current record's timestamp
`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()`	`org.apache.kafka.common.header.Headers`
`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 exist in the topology and be connected to the processor. Two ways to do that:

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 refer to it by name from the 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,
    ),
)

val out = events.process({ SessionProcessor("sessions") }, Named.`as`("sessionize"))

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

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

KStream<String, Event> out =
    events.process(() -> new SessionProcessor("sessions"), Named.as("sessionize"));

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.

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

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"))
        .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"))
            .to("counts-out", Produced.<String, Long>as("sink").withValueSerde(Serdes.Long()));
    }
}

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.