StoatFlow

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 restartYes — restored from local disk + changelogNo — rebuilt from changelog on every start
Memory footprintBounded; data spills to diskWhole store lives on the heap
Exactly-onceYesYes (changelog still committed on the barrier)
Best forProduction, large state, fast restartTests, 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 typePersistent (RocksDB)In-memory
Key-valuepersistentKeyValueStore(name)inMemoryKeyValueStore(name)
LRU cache—lruMap(name, maxCacheSize)
WindowpersistentWindowStore(name, retention, windowSize)inMemoryWindowStore(name, retention, windowSize)
SessionpersistentSessionStore(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)

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

Store type precedence

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

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

Materialized options

MethodEffect
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.
withRecordHeaders() / withoutRecordHeaders()Persists the processing record's Headers alongside the value (KIP-1271/KIP-1285). See record headers in state stores. Explicit opt-in or opt-out always beats the global stoatflow.dsl.store-format.
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()),
        )
}

Attaching a store to a custom Processor

For the Processor API you create a store builder yourself, register it on the StreamsBuilder with addStateStore(...), and connect it to the processor by naming it in the process(...) call — registering alone is not enough. Inside the processor you then look it up with context.getStateStore(name). The builder comes from Stores.keyValueStoreBuilder(supplier, keySerde, valueSerde) (and the windowStoreBuilder / sessionStoreBuilder / versionedKeyValueStoreBuilder equivalents).

The alternative is to declare the store from ProcessorSupplier.stores(), which connects it automatically — see Processor API.

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") }, "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)
    }
}
A processor may only reach a store it declared.getStateStore(name) throws StoreNotConnectedException if name is not connected to that node — the Kafka Streams rule, and a fatal error rather than something processing.exception.handler can skip. Two exemptions need no declaration: global stores (addGlobalStore(...) and globalTable(...)) and read-only stores (addReadOnlyStateStore(...)).

A global store reached without declaring it is handed to the processor read-only — put/delete throw UnsupportedOperationException, as in Kafka Streams. Its own state-update processor writes to it normally. This covers every store family: plain key-value, window and session stores, and equally the timestamped, versioned and headers-aware ones. The store keeps its declared type, so getStateStore<TimestampedKeyValueStore<K, V>>(name) still returns a TimestampedKeyValueStore; only its writes throw.

To write to a global store from your own processor, declare it — process(supplier, "my-global") or ProcessorSupplier.stores() — which opts into write access. That is a StoatFlow extension (Kafka Streams rejects the declaration itself): on a single instance all state is already global, so a "global" store is an ordinary store, and the read-only default exists for Kafka Streams portability rather than safety.

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 — atomic within a commit epoch. See State and thread safety for the full model, including the cross-key key-lock utility and the two shapes where key affinity does not hold: a many-to-one re-key into a Processor API node, which StoatFlow refuses to compile rather than let you lock your way around, and storing under a key that is not the record key, which is computed at runtime and so cannot be caught at build time.

Record headers in state stores

A store can persist the processing record's Headers alongside the value — a schema id, a traceparent, tenant or routing metadata, lineage — so it survives aggregation and comes back through interactive queries and on the changelog. This is Kafka's KIP-1271 (storage) and KIP-1285 (DSL opt-in), and StoatFlow implements both.

Opt in per store, or globally:

builder
    .stream<String, String>("orders")
    .groupByKey()
    .count(
        Materialized.`as`<String, Long, StateStore>("order-counts")
            .withKeySerde(Serdes.String())
            .withValueSerde(Serdes.Long())
            .withRecordHeaders(),
    )

stoatflow:
  dsl:
    store-format: HEADERS   # every DSL aggregation persists headers by default

Headers-aware stores are timestamped-only (as in Kafka Streams), covering timestamped key-value, versioned, timestamped window and session stores. Join buffers are out of scope in both engines. Query them with the QueryableStoreTypes.*WithHeaders() views.

Turning record headers on

Flipping the flag on is a lazy in-place upgrade — no migration pause, no I/O spike. The store opens with a second RocksDB keyspace: reads check the new one and fall back to the old, converting an old-format hit on the way out (old entries read back with empty headers, which is correct — they predate the feature); writes land in the new keyspace and clear the old copy, so each key upgrades itself the first time it is touched. Windowed and session stores do this per time segment.

Turning record headers off again

Turning them off is not the mirror image, and it is worth knowing before you flip the flag.

The new keyspace is created eagerly the first time the store opens in headers mode, so it exists on disk even if nothing was ever written to it — and RocksDB refuses to open a database without naming every keyspace present. StoatFlow therefore classifies the mismatch at startup, before the store opens, and treats it as a declared format change, not corruption:

SituationWhat happens
Nothing was written in headers modeThe empty keyspace is dropped in place. Free: no acknowledgment, no restore, not a byte of data touched.
Data has migratedStartup refuses with an actionable message. Nothing on disk is touched, so an accidental flag removal costs a restart, never data.
Data has migrated, and you acknowledgeSet stoatflow.state.format-downgrade: wipe-and-restore. That store's local state is deleted and rebuilt from its changelog — one full restore, bounded by changelog size.
Data has migrated, and the store has no recovery sourceRefused unconditionally. The acknowledgment does not apply: with the changelog disabled (per store via withLoggingDisabled(), or globally) there is nothing to rebuild from, so wiping would be unrecoverable data loss.
A downgrade sheds persisted headers, and turning the flag back on does not bring them back. The old format cannot hold headers, so they are dropped from the local store; and re-enabling headers upgrades lazily, converting legacy values with empty headers. Only a forced full restore repacks them. Kafka Streams behaves the same way — it refuses the downgrade outright and leaves you to delete local state by hand; the acknowledgment key is StoatFlow's addition, so the operation is one config line rather than a manual reset.

The outcome is metered as stoatflow.store.format.downgrade.total{outcome=dropped-empty|wiped|refused}. If you take the rebuild, the restoration tuning guide covers making it fast.

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 kindQueryableStoreTypes factoryRead-only view
Key-valuekeyValueStore()ReadOnlyKeyValueStore<K, V>
WindowwindowStore()ReadOnlyWindowStore<K, V>
SessionsessionStore()ReadOnlySessionStore<K, AGG>
Timestamped KVtimestampedKeyValueStore()ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>
Timestamped windowtimestampedWindowStore()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.

Range bounds follow Kafka Streams semantics (KIP-763): a null bound is open-ended (both null ≡ all()), reverseRange takes its bounds in the same ascending (low, high) order as range, and an inverted range (from > to on the serialized key bytes) returns an empty iterator with a WARN. The same null-bound rules apply to ReadOnlyWindowStore.fetch(keyFrom, keyTo, …) and the ReadOnlySessionStore key-range queries.

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

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)

Related

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.