State stores

Use the Stores factory and Materialized to configure persistent and in-memory key-value, window, session, and versioned stores — and read them back via interactive queries.

State stores hold the data your stateful operators accumulate — running counts, window aggregates, join buffers, custom per-key state. This page covers the two factories you use to configure them (Stores and Materialized), how to choose between RocksDB-backed and in-memory variants, how to attach a store to a custom Processor, and how to read a store back from a running application.

For the conceptual model — global state, thread-safety under concurrent lanes, and changelog-backed durability — see State and thread safety.

RocksDB vs. in-memory

Every store type comes in two variants, and the choice is the same each time:

	Persistent (RocksDB)	In-memory
Survives restart	Yes — restored from local disk + changelog	No — rebuilt from changelog on every start
Memory footprint	Bounded; data spills to disk	Whole store lives on the heap
Exactly-once	Yes	Yes (changelog still committed on the barrier)
Best for	Production, large state, fast restart	Tests, small bounded state, prototyping

RocksDB is the default and the recommended production type. In-memory stores keep their data on the JVM heap, so they're bounded by available memory and have to read the entire changelog at startup — fine for tests and small state, but RocksDB recovers faster and scales past heap size in production. Note that "in-memory store" is not "no durability": both variants write a changelog and commit it atomically on the commit barrier. The difference is local persistence and cold-start cost.

The `Stores` factory

Stores is a static factory (io.stoatflow.core.state.Stores) that returns a supplier — a deferred description of a store. Suppliers are handed to Materialized (for DSL operators) or to StreamsBuilder.addStateStore(...) (for custom Processors). The factory methods, grounded in Stores.kt:

Store type	Persistent (RocksDB)	In-memory
Key-value	`persistentKeyValueStore(name)`	`inMemoryKeyValueStore(name)`
LRU cache	—	`lruMap(name, maxCacheSize)`
Window	`persistentWindowStore(name, retention, windowSize)`	`inMemoryWindowStore(name, retention, windowSize)`
Session	`persistentSessionStore(name, retention, inactivityGap)`	`inMemorySessionStore(name, retention, inactivityGap)`
Timestamped KV (KIP-258)	`persistentTimestampedKeyValueStore(name)`	`inMemoryTimestampedKeyValueStore(name)`
Timestamped window (KIP-258)	`persistentTimestampedWindowStore(name, retention, windowSize)`	`inMemoryTimestampedWindowStore(name, retention, windowSize)`
Versioned KV (KIP-889)	`persistentVersionedKeyValueStore(name, historyRetention)`	`inMemoryVersionedKeyValueStore(name, historyRetention)`

All duration arguments are java.time.Duration. Window and session retention must be at least the window size (the factory enforces this and throws IllegalArgumentException otherwise). Store names must be non-empty and contain only [a-zA-Z0-9._-].

Versioned stores (persistentVersionedKeyValueStore / inMemoryVersionedKeyValueStore) keep multiple versions of each value over time for point-in-time lookups and temporal joins. The single historyRetention argument is dual-purpose: it's both how long old versions are kept and the grace period — out-of-order records older than that window are rejected. Unlike Kafka Streams, which only ships a RocksDB-backed versioned store, StoatFlow offers an in-memory variant too.

`Materialized` — stores behind DSL operators

Stateful DSL operators (count, reduce, aggregate, windowed counts, joins) take a Materialized to describe their store. The most common forms (grounded in Materialized.kt):

import io.stoatflow.core.state.StateStore
import io.stoatflow.core.state.StoreType
import io.stoatflow.core.topology.Materialized
import org.apache.kafka.common.serialization.Serdes

// Named store, default (RocksDB) type, with serdes
val counts =
    Materialized.`as`<String, Long, StateStore>("word-counts")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long())

// Pick the store type explicitly (clears any explicit supplier)
val inMem =
    Materialized.`as`<String, Long, StateStore>("counts")
        .withStoreType(StoreType.IN_MEMORY)

// Auto-generated store name, store type only
val autoNamed = Materialized.`as`<String, Long, StateStore>(StoreType.ROCKS_DB)

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.state.StoreType;
import io.stoatflow.core.topology.Materialized;
import org.apache.kafka.common.serialization.Serdes;

// Named store, default (RocksDB) type, with serdes
var counts =
    Materialized.<String, Long, StateStore>as("word-counts")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long());

// Pick the store type explicitly (clears any explicit supplier)
var inMem =
    Materialized.<String, Long, StateStore>as("counts")
        .withStoreType(StoreType.IN_MEMORY);

// Auto-generated store name, store type only
var autoNamed = Materialized.<String, Long, StateStore>as(StoreType.ROCKS_DB);

You can also hand Materialized.as(...) a supplier directly when you need parameters the DSL doesn't expose (window retention overrides, an explicit window/session/versioned supplier):

import io.stoatflow.core.state.Stores
import io.stoatflow.core.topology.Materialized
import java.time.Duration

// Explicit persistent window store supplier
val windowed =
    Materialized.`as`<String, Long, StateStore>(
        Stores.persistentWindowStore("hourly", Duration.ofHours(2), Duration.ofHours(1)),
    )

// Explicit in-memory key-value supplier
val cache =
    Materialized.`as`<String, Long, StateStore>(Stores.inMemoryKeyValueStore("cache"))

import io.stoatflow.core.state.Stores;
import io.stoatflow.core.topology.Materialized;
import java.time.Duration;

// Explicit persistent window store supplier
var windowed =
    Materialized.<String, Long, StateStore>as(
        Stores.persistentWindowStore("hourly", Duration.ofHours(2), Duration.ofHours(1)));

// Explicit in-memory key-value supplier
var cache =
    Materialized.<String, Long, StateStore>as(Stores.inMemoryKeyValueStore("cache"));

Store type precedence

When more than one store source is set on a Materialized, resolution follows a fixed order (ADR-029):

withStoreType(...) (the DslStoreSuppliers / StoreType) — wins absolutely, and clears any explicit supplier.
An explicit supplier passed to Materialized.as(supplier).
Default — RocksDB (matching Kafka Streams).

`Materialized` options

Method	Effect
`withKeySerde(serde)` / `withValueSerde(serde)`	Serdes for the store (and its changelog). Falls back to the configured defaults if unset.
`withStoreType(StoreType.IN_MEMORY \| ROCKS_DB)`	Selects store type; takes precedence over any explicit supplier.
`withRetention(duration)`	Overrides retention for windowed stores. Must be at least window size + grace.
`withCachingEnabled()` / `withCachingEnabled(config)`	Compacts multiple writes to the same key within a barrier interval; only the final value per key is emitted downstream via `toStream()`.
`withCachingDisabled()`	Emits every intermediate update downstream.
`withLoggingEnabled()` / `withLoggingEnabled(topicConfig)`	Forces the changelog topic on, optionally with custom topic config (e.g. `retention.ms`).
`withLoggingDisabled()`	Disables the changelog for this store — state is then only locally persisted (RocksDB) and cannot be recovered from Kafka after a crash. Use only for state you can recompute.

Caching only affects what flows downstream. Whether or not caching is enabled, changelog writes are always compacted to the final value per key per barrier interval. withCachingDisabled() is what you want when a downstream consumer needs to observe every intermediate update (as the word-count walkthrough shows — counts climbing 1, 2, 3 rather than only the final tally). When unset, caching follows the global default.

Worked example: a counting KTable

import io.stoatflow.core.state.StateStore
import io.stoatflow.core.topology.Grouped
import io.stoatflow.core.topology.Materialized
import io.stoatflow.core.topology.Named
import io.stoatflow.core.topology.StreamsBuilder
import org.apache.kafka.common.serialization.Serdes

fun buildTopology(builder: StreamsBuilder) {
    builder
        .stream<String, String>("words")
        .groupBy({ _, word -> word }, Grouped.`as`("group-by-word"))
        .count(
            Named.`as`("count"),
            Materialized.`as`<String, Long, StateStore>("word-counts")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.Long()),
        )
}

import io.stoatflow.core.state.StateStore;
import io.stoatflow.core.topology.Grouped;
import io.stoatflow.core.topology.Materialized;
import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.StreamsBuilder;
import org.apache.kafka.common.serialization.Serdes;

static void buildTopology(StreamsBuilder builder) {
    builder
        .<String, String>stream("words")
        .groupBy((key, word) -> word, Grouped.as("group-by-word"))
        .count(
            Named.as("count"),
            Materialized.<String, Long, StateStore>as("word-counts")
                .withKeySerde(Serdes.String())
                .withValueSerde(Serdes.Long()));
}

Attaching a store to a custom Processor

For the Processor API you create a store builder yourself and register it on the StreamsBuilder with addStateStore(...), then look it up by name inside the processor via context.getStateStore(name). The builder comes from Stores.keyValueStoreBuilder(supplier, keySerde, valueSerde) (and the windowStoreBuilder / sessionStoreBuilder / versionedKeyValueStoreBuilder equivalents).

import io.stoatflow.core.processor.Processor
import io.stoatflow.core.processor.ProcessorContext
import io.stoatflow.core.processor.Record
import io.stoatflow.core.state.KeyValueStore
import io.stoatflow.core.state.Stores
import io.stoatflow.core.topology.StreamsBuilder
import org.apache.kafka.common.serialization.Serdes

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

    builder
        .stream<String, String>("events")
        .process({ CountingProcessor("seen-counts") })
}

class CountingProcessor(
    private val storeName: String,
) : Processor<String, String, String, Long> {
    private lateinit var store: KeyValueStore<String, Long>

    override fun init(context: ProcessorContext<String, Long>) {
        store = context.getStateStore(storeName)
    }

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

import io.stoatflow.core.processor.Processor;
import io.stoatflow.core.processor.ProcessorContext;
import io.stoatflow.core.processor.Record;
import io.stoatflow.core.state.KeyValueStore;
import io.stoatflow.core.state.Stores;
import io.stoatflow.core.topology.StreamsBuilder;
import org.apache.kafka.common.serialization.Serdes;

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

    builder
        .<String, String>stream("events")
        .process(() -> new CountingProcessor("seen-counts"));
}

class CountingProcessor implements Processor<String, String, String, Long> {
    private final String storeName;
    private KeyValueStore<String, Long> store;

    CountingProcessor(String storeName) {
        this.storeName = storeName;
    }

    @Override
    public void init(ProcessorContext<String, Long> context) {
        store = context.getStateStore(storeName);
    }

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

Within a processor, state access is safe without any locking of your own: records with the same key always run on the same processing context in arrival order, so per-key read-modify-write is serial. The store also exposes atomic compute(key, fn) and merge(key, value, fn) for read-modify-write in a single call. See State and thread safety for the full model, including the cross-key key-lock utility.

Reading a store back

A store registered in your topology is queryable by name. The read-only views are typed via QueryableStoreTypes and selected with StoreQueryParameters.fromNameAndType(...):

Store kind	`QueryableStoreTypes` factory	Read-only view
Key-value	`keyValueStore()`	`ReadOnlyKeyValueStore<K, V>`
Window	`windowStore()`	`ReadOnlyWindowStore<K, V>`
Session	`sessionStore()`	`ReadOnlySessionStore<K, AGG>`
Timestamped KV	`timestampedKeyValueStore()`	`ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>`
Timestamped window	`timestampedWindowStore()`	`ReadOnlyWindowStore<K, ValueAndTimestamp<V>>`

ReadOnlyKeyValueStore gives you get(key), containsKey(key), all(), reverseAll(), range(from, to), reverseRange(from, to), prefixScan(prefix, prefixKeySerializer), and approximateNumEntries(). Iterators must be closed after use.

StoatFlow's single-instance model means all state is globally accessible — there is no partition routing and no cross-instance RPC. Consequently StoreQueryParameters has no withPartition(...); a query against any key resolves locally. Use enableStaleStores() to allow reads while the application is still restoring.

Interactive queries against the running engine

The query entry point is store(...) on the core StoatFlow engine, which returns the requested read-only view:

import io.stoatflow.core.state.QueryableStoreTypes
import io.stoatflow.core.state.ReadOnlyKeyValueStore
import io.stoatflow.core.state.StoreQueryParameters

val store: ReadOnlyKeyValueStore<String, Long> =
    stoatflow.store(
        StoreQueryParameters.fromNameAndType(
            "word-counts",
            QueryableStoreTypes.keyValueStore(),
        ),
    )

val count: Long? = store.get("hello")

import io.stoatflow.core.state.QueryableStoreTypes;
import io.stoatflow.core.state.ReadOnlyKeyValueStore;
import io.stoatflow.core.state.StoreQueryParameters;

ReadOnlyKeyValueStore<String, Long> store =
    stoatflow.store(
        StoreQueryParameters.fromNameAndType(
            "word-counts",
            QueryableStoreTypes.keyValueStore()));

Long count = store.get("hello");

store(...) throws IllegalStateException unless the application is RUNNING (or RESTORING / VALIDATING_STATE when enableStaleStores() was set), IllegalArgumentException if the named store does not exist, and IllegalArgumentException if the store's type does not match the requested QueryableStoreType. storeNames() lists every registered store name for discovery.

store(...) and storeNames() are defined on the core StoatFlow engine, so programmatic interactive queries need a handle to that engine. A :core application holds it directly. The :runtime wrapper (StoatFlowRuntime.fromConfig) currently keeps its engine instance private and exposes no public query accessor, so typed in-process interactive queries are not available from a :runtime deployment today — use the :core API directly if you need them.

Reading a store in a test

In TopologyTestDriver (the in-memory test driver), read stores back directly with getKeyValueStore(name) (and getWindowStore / getSessionStore). Call commitBarrier() first if the store has caching enabled, so cached emissions are flushed:

import io.stoatflow.testutils.TopologyTestDriver

val driver = TopologyTestDriver.fromBuilder(builder)
// ... pipe input ...
driver.commitBarrier() // flush cached writes before reading

val store = driver.getKeyValueStore<String, Long>("word-counts")
assertThat(store.get("the")).isEqualTo(3L)

import io.stoatflow.testutils.TopologyTestDriver;
import io.stoatflow.core.state.KeyValueStore;

var driver = TopologyTestDriver.fromBuilder(builder);
// ... pipe input ...
driver.commitBarrier(); // flush cached writes before reading

KeyValueStore<String, Long> store = driver.getKeyValueStore("word-counts");
assertThat(store.get("the")).isEqualTo(3L);

State and thread safety — global state, concurrent-access correctness, durability
Aggregations and Windowing — operators that materialize stores
Serdes — the serialization the store uses for keys, values, and changelog
Testing — reading stores back from the test driver
Kafka Streams compatibility matrix — store-type parity against Kafka Streams

Error handling and DLQ

Configure deserialization, processing, and production exception handlers — log-and-continue, log-and-fail, or dead-letter-queue — in code or YAML, and inspect the DLQ headers StoatFlow writes.

Testing topologies

Unit-test a StoatFlow topology in-memory with TopologyTestDriver — pipe records, advance time and watermarks, read state stores, and drive YAML config through StoatFlowTestDriver. No broker required.