State and thread-safety

State in StoatFlow is global: every state store lives in the single JVM running your topology, and any processing lane can read or write any key. This page explains why that's safe under concurrent access, the one case where you have to coordinate it yourself, and the store types you can choose from.

This is the canonical reference for the thread-safety model. For the engine-level picture of lanes, the commit barrier, and durability, see Architecture. For the how-to — declaring stores, choosing types, and accessing them from the DSL or a custom Processor — see State stores.

The global state model

In the standard Kafka Streams model, state is partitioned: a task owns a slice of the keyspace, and its state stores hold only the keys for the partitions that task is assigned. Reading a key that lives on another task means an interactive-query round-trip to another instance.

StoatFlow has no partitions to scope state to. Because the application runs as exactly one instance, every state store holds the entire keyspace, in one process. Three things follow:

• Any lane can read or write any key. There is no "this key belongs to that task" boundary — a lane processing one record can look up a value written by a record on a completely different key.
• No replication of the same data across JVMs. A single process holds all state; there's no copy of the same store living on another node to keep consistent.
• No inter-instance lookup protocol. A value is either in the local store or it doesn't exist yet — there's never a network hop to find it.

Durability is unchanged from the Kafka Streams approach: every write produces a Kafka changelog entry, and on restart the runtime rebuilds local state from the changelog. That side of the story — changelog coupling to the commit barrier and parallel restoration — is on Architecture.

Why concurrent access is safe

Global state plus many lanes running in parallel sounds like it should require locking around every store access. It doesn't, and the reason is key affinity (see Lanes and parallelism).

Records are routed to lanes by their key. The same key always lands on the same lane, every time. So for any single key:

• All updates to that key are processed serially, by one lane, in the order Kafka delivered them.
• There is never a second lane writing the same key concurrently.

And across keys:

• Different keys live on different lanes and update in parallel.
• Because they're different keys, those updates touch different store entries — no contention.

The net effect is per-key serialization without a global store lock. You get the ordering guarantee you'd expect from single-threaded processing on each key, and the parallelism of many lanes across the keyspace. Reading and writing a state store from inside an operator — count, aggregate, a join, or your own Processor — needs no synchronization from you for the common single-key case.

Here, a custom Processor reads the current value for the record's key, updates it, and writes it back. Because the record's key pins it to one lane, this read-modify-write is already serial for that key — no locking required:

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 RunningSumProcessor(
    private val storeName: String,
) : ContextualProcessor<String, Long, String, Long>() {
    private lateinit var store: KeyValueStore<String, Long>

    override fun init(context: ProcessorContext<String, Long>) {
        super.init(context)
        // storeName must be connected to this node — declare it from ProcessorSupplier.stores(),
        // or name it at the call site: process(supplier, storeName).
        store = context.getStateStore(storeName)
    }

    override fun process(record: Record<String, Long>) {
        // Single-key read-modify-write: safe with no locking,
        // because this key is always handled by the same lane.
        val current = store.get(record.key()) ?: 0L
        val updated = current + record.value()
        store.put(record.key(), updated)
        context().forward(record.withValue(updated))
    }
}

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 RunningSumProcessor
        extends ContextualProcessor<String, Long, String, Long> {

    private final String storeName;
    private KeyValueStore<String, Long> store;

    public RunningSumProcessor(String storeName) {
        this.storeName = storeName;
    }

    @Override
    public void init(ProcessorContext<String, Long> context) {
        super.init(context);
        // storeName must be connected to this node — declare it from ProcessorSupplier.stores(),
        // or name it at the call site: process(supplier, storeName).
        this.store = context.getStateStore(storeName);
    }

    @Override
    public void process(Record<String, Long> record) {
        // Single-key read-modify-write: safe with no locking,
        // because this key is always handled by the same lane.
        long current = store.get(record.key()) == null ? 0L : store.get(record.key());
        long updated = current + record.value();
        store.put(record.key(), updated);
        context().forward(record.withValue(updated));
    }
}

Custom processors: instance fields are per-lane

A custom Processor (or FixedKeyProcessor) is instantiated once per lane — a lane is StoatFlow's analog of a Kafka Streams task, so this mirrors KS's one-instance-per-task model. Your instance fields are therefore per-lane and single-threaded: safe by construction for per-key scratch state, with no synchronization needed, exactly as in KS.

Two things follow from per-lane instances:

• A punctuator fires once over global state on the punctuation lane — it runs against a different instance than your record lanes, so it cannot see instance fields set in process(...). For a scheduled write to a key, use a timer (onTimer), which runs in that key's lane with full read/write access.
• For state that must be shared across lanes or read from a punctuator, put it in a state store, not an instance field — that's exactly what the global state store is for. Punctuators have full read/write store access. For ephemeral scratch that needn't survive a restart, an in-memory store with withLoggingDisabled() is the lightweight option. See Attaching state stores.

The one case that needs coordination: cross-key atomic updates

Key affinity serializes access per key. It says nothing about two different keys handled by two different lanes at the same time. That's almost always fine — different keys are independent. But there's one pattern it doesn't cover: a custom Processor that needs to read-modify-write more than one key as a single atomic step, where another lane might be touching one of those same keys concurrently.

Examples: moving a balance from account A to account B (debit one, credit the other, with no moment where the total is wrong), or maintaining an invariant that spans a pair of related keys. Because A and B may hash to different lanes, two records could enter the critical section at the same time, and the interleaving would corrupt the invariant.

For this — and only this — StoatFlow provides a key-lock utility, KeyLockManager (io.stoatflow.core.state.KeyLockManager). You instantiate and own it; the runtime never creates one for you. It guards a short critical section across one or more keys, acquiring locks in a deterministic order so concurrent multi-key sections can't deadlock against each other.

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

class TransferProcessor(
    private val storeName: String,
) : ContextualProcessor<String, Transfer, String, Transfer>() {
    private lateinit var balances: KeyValueStore<String, Long>

    // One instance per store; share one across stores for cross-store atomicity.
    private val keyLocks = KeyLockManager()

    override fun init(context: ProcessorContext<String, Transfer>) {
        super.init(context)
        // Connected at the call site: process(supplier, storeName).
        balances = context.getStateStore(storeName)
    }

    override fun process(record: Record<String, Transfer>) {
        val t = record.value()
        // Both keys may live on different lanes — lock both for the atomic move.
        keyLocks.withLocks(t.from, t.to) {
            val fromBalance = balances.get(t.from) ?: 0L
            val toBalance = balances.get(t.to) ?: 0L
            balances.put(t.from, fromBalance - t.amount)
            balances.put(t.to, toBalance + t.amount)
        }
        context().forward(record)
    }
}

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

public class TransferProcessor
        extends ContextualProcessor<String, Transfer, String, Transfer> {

    private final String storeName;
    private KeyValueStore<String, Long> balances;

    // One instance per store; share one across stores for cross-store atomicity.
    private final KeyLockManager keyLocks = new KeyLockManager();

    public TransferProcessor(String storeName) {
        this.storeName = storeName;
    }

    @Override
    public void init(ProcessorContext<String, Transfer> context) {
        super.init(context);
        // Connected at the call site: process(supplier, storeName).
        this.balances = context.getStateStore(storeName);
    }

    @Override
    public void process(Record<String, Transfer> record) {
        Transfer t = record.value();
        // Both keys may live on different lanes — lock both for the atomic move.
        keyLocks.withLocks(new Object[]{ t.from, t.to }, () -> {
            long fromBalance = balances.get(t.from) == null ? 0L : balances.get(t.from);
            long toBalance = balances.get(t.to) == null ? 0L : balances.get(t.to);
            balances.put(t.from, fromBalance - t.amount);
            balances.put(t.to, toBalance + t.amount);
            return null;
        });
        context().forward(record);
    }
}

If a lock can't be acquired within the configured timeout (default 10 seconds), the call throws KeyLockTimeoutException (io.stoatflow.core.exception) rather than blocking forever — a loud signal of a deadlock or excessive contention rather than a silent stall.

Store types

StoatFlow ships five store types. Stateful DSL operators (count, reduce, aggregate, joins, windowed and session aggregations, suppress) pick the appropriate type automatically; custom Processors declare the stores they need. Each type comes in two backends:

• RocksDB-backed — persistent, on-disk. The default. State larger than memory is fine; local data survives process restarts and gives the changelog a head-start on recovery.
• In-memory — held entirely in the JVM heap. Lower per-operation overhead; bounded by available memory. Still durable via the changelog — the in-memory store is rebuilt from it on restart.

The backend is chosen per store. The DSL uses RocksDB by default; you switch a materialized store to in-memory with Materialized.withStoreType(StoreType.IN_MEMORY), and custom processors choose via the Stores factory. The full how-to — declaring stores, the precedence rules, and the factory API — is on State stores.

Where to go next

• State stores — declaring stores, choosing types, and accessing them from the DSL or a Processor
• Lanes and parallelism — the key-affinity routing that makes concurrent state access safe
• Architecture — the single-instance engine, commit barrier, and changelog durability
• Processor API — building custom processors that own their state