State stores

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:

Which timestamp a restored table holds. A store rebuilt from its changelog gets back exactly the timestamps it stored. A table() store rebuilt from its source topic instead — a reused compacted source, Consumed.materializeFromSourceTopic(true), or a read-only store (addReadOnlyStateStore, including the overload without an update processor, which writes the extracted timestamp live) — takes each record's Kafka record timestamp, as Kafka Streams does. A global store (globalTable, addGlobalStore) is rebuilt from its changelog and keeps the timestamps it stored. If the table uses a custom TimestampExtractor or timestamp assigner, the restored entries therefore carry the record timestamp rather than the extracted one, and output timestamps derived from them (joins against the table, aggregations over it) can differ after a restart or an HA failover until each key is updated again.

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:

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._-].

Window and session retention

Two things scope that bound, both covered in Tuning: under EmitStrategy.onWindowClose() expiry is gated on the durable emit frontier rather than the watermark, which adds an unbounded (watermark − emit frontier) term — and because that gate lives in the processor, it delays in-memory expiry too, so the exactly-retention property above is an onWindowUpdate() statement. And after a retention narrowing, existing segments keep the wider span they were written with, so the ceiling is retention + widest existing span until they age out.

Which term sets the interval — and therefore how much slack you get — depends on the configuration:

A multiplier only means something with the condition that makes it bind. Two consequences catch people out: below three minutes of retention the 60-second floor does not always govern (a 90-second window inside a 2-minute retention makes the window-size term the largest, so the store holds two segments), and the retention + 60s regime is not Kafka-Streams-only — StoatFlow has had the same floor since ADR-141, so a 30-second session store keeps 3× what you configured.

The mechanism: a segment is dropped only once its entire time range has fallen at or before watermark − retention. The segment containing the boundary is retained in full — that is the slack. Worked examples for every regime are on the windowing page, and sizing guidance is in Tuning.

LRU stores

Stores.lruMap(name, maxCacheSize) keeps at most maxCacheSize keys and evicts the least recently used one — a get() counts as a use. That makes it a different kind of store from every other key-value store, and it has trade-offs worth knowing before you materialise a table with it:

• No commit buffer. Other stores buffer each commit's writes and apply them when the commit succeeds; an LRU store writes directly (a buffer cannot sit above a store that decides its own evictions). It remains exactly-once — an aborted commit ends in an engine rebuild from the changelog, so its writes never survive (with logging disabled the rebuild is empty, so the store loses all its state on every such restart) — but it cannot be read as of the last commit, so under exactly_once it cannot be a table that a join reads from another lane (joins under exactly-once), and a Processor API node that may read it from another lane gets the presumed-cross-lane-read warning. Its reads are always live: when read-isolation levels arrive (KIP-892), it will not honour READ_COMMITTED (KSC-116).
• Reads change it. A get() — including an interactive query — promotes the key, so it changes which key is evicted next.
• A restart does not bring back the same keys. It is rebuilt from its changelog, and when the changelog holds more than maxCacheSize keys, which of them survive depends on the restore order — the changelog's partitions are restored in parallel — not on which keys were most recently used. The surviving set can even differ between two restarts over the same data, so a table backed by it can answer differently after a restart or an HA failover.
• Evictions are not in the changelog by default. An evicted key is dropped locally only, so the changelog keeps its last value. Opt in with Stores.lruMap(name, maxCacheSize, tombstoneOnEviction = true) (Java: Stores.lruMap(name, maxCacheSize, true)): each eviction then also writes a delete to the changelog, so a restart or failover does not bring an evicted key back. It needs changelog logging — combined with withLoggingDisabled() the store is refused at startup.

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 Kafka Streams' order:

1. An explicit supplier passed to Materialized.as(supplier) — wins.
2. withStoreType(...) (the DslStoreSuppliers / StoreType) — applies only when no explicit supplier is set.
3. Default — RocksDB, segmented for window and session stores (matching Kafka Streams).

Neither setter clears the other. If both are set, the supplier is used and StoatFlow logs one warning per store naming the ignored store type. The same order applies per side on StreamJoined (withThisStoreSupplier / withOtherStoreSupplier beat withDslStoreSuppliers), and a foreign-key join's subscription store follows the backend of the result store's supplier. Before 2026-09-09 this was inverted — the store type outranked and erased an explicit supplier — so a ported topology that set both silently changed backend (KSC-100 in the compatibility matrix).

Materialized options

Changelog partition count

Each store's changelog topic has a partition count. By default every store uses stoatflow.changelog.num-partitions (3), and withChangelogNumPartitions(n) overrides it for one store:

// A large, write-hot aggregation store: more partitions, more write throughput and leader spread.
Materialized.`as`<String, Stats>("order-stats")
    .withChangelogNumPartitions(6)

// A small lookup table: pin it back down so it stops widening every commit.
Materialized.`as`<String, Config>("feature-flags")
    .withChangelogNumPartitions(1)

The same method is on StreamJoined (for a join's buffer stores), on the Processor-API StoreBuilder, and on Suppressed.BufferConfig. The timer store has no Materialized, so it uses the stoatflow.changelog.timer-num-partitions config key instead (-1 inherits the global count).

Two stores you cannot name inherit the setting from one you can: a stream-stream LEFT/OUTER join's -outer-store inherits from StreamJoined, and a foreign-key join's -subscription-store inherits from the result store's Materialized. Both hold roughly one entry per row of the store they shadow, so sizing the visible store sizes its hidden half too.

What the number buys. A changelog topic's partition count sets its write-throughput ceiling and decides how the store's data and write load shard across brokers — one partition funnels every write for that store through a single broker leader. Cold-start restore also gets faster, because the restoration pool creates one task per changelog partition — but that gain is sub-linear: a single store restores into one RocksDB instance whose writes serialise internally, and all stores share one pool of stoatflow.state.restoration-pool-size workers.

What it costs. Changelog records ride the same Kafka transaction as your output records. A store whose epoch touches keys spanning all N partitions therefore adds up to N−1 partitions to every commit that touches it — more produce requests, more commit markers, a wider flush wait — plus extra standby consumer assignments and one more local offset write per partition per commit. That cost is paid continuously; the restore benefit is paid at a cold start you hope is rare.

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

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

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.

Deleting a key

Two calls delete a key from a KeyValueStore, and they do not cost the same:

• store.put(key, null) writes the tombstone and nothing else.
• store.delete(key) returns the previous value, so it has to read — and deserialise — that value before it writes the same tombstone. That is the Kafka Streams contract, and StoatFlow keeps it.

Use put(key, null) whenever you do not need the value you are deleting. The read that delete() adds is nearly free for a key written in the current commit epoch (it is still in the write-back cache), but for anything older it is a full store read — and expiry, session close-out and cleanup work, where most deletes happen, almost always delete keys written many epochs ago. StoatFlow's own DSL tombstones through put(key, null) for exactly this reason, and so does Kafka Streams.

override fun onTimer(timestamp: Long, key: String, context: TimerContext<String, Session>) {
    val sessions = context.getStateStore<KeyValueStore<String, Session>>("sessions")
    val session = sessions.get(key) ?: return
    if (session.lastActivity + SESSION_TIMEOUT_MS <= timestamp) {
        context.forward(key, session.complete())
        sessions.put(key, null) // tombstone; we already read the value above
    }
}

@Override
public void onTimer(long timestamp, String key, TimerContext<String, Session> context) {
    KeyValueStore<String, Session> sessions = context.getStateStore("sessions");
    Session session = sessions.get(key);
    if (session != null && session.lastActivity() + SESSION_TIMEOUT_MS <= timestamp) {
        context.forward(key, session.complete());
        sessions.put(key, null); // tombstone; we already read the value above
    }
}

On a timestamped store a null value is an absent entry, as in Kafka Streams. Writing a ValueAndTimestamp whose value is null — or one whose value serializer returns null — deletes the key (on a window store, the window); to delete, write put(key, null) (from Java, put(key, ValueAndTimestamp.makeAllowNullable(null, ts)) does the same). merge rejects such a value with IllegalArgumentException, as Map.merge rejects a null — use put(key, null) to delete — while a null returned by the remapping function deletes the key. putIfAbsent(key, null) writes nothing. put and putIfAbsent accept null from Kotlin and Java alike, as Kafka Streams' do (merge does not — like java.util.Map.merge, it needs a value), and a window store's put(key, null, windowStart) deletes the window (a no-op on a store that retains duplicates). On a session store, put(session, null) removes the session. Reads mirror it: a stored value that your deserializer maps to null reads back as null, and an iterator yields it as a KeyValue whose value is null. The factories follow Kafka Streams too: ValueAndTimestamp.make(null, ts) returns null, and makeAllowNullable(null, ts) returns an instance holding the null.

Expiring sessions from an in-memory store

The in-memory session store keeps its sessions in one key-first map, which is what makes the per-record session merge scan a tight seek. Expiry is the one operation with no key, so by default it walks every live session on every watermark advance — about 50 ns per session in StoatFlow's reference measurement: 25 ms per barrier at 500 000 live sessions, and past roughly 3 million sessions per store the pass alone costs more than the 150 ms barrier floor. (The per-session cost depends on heap layout — the same pass measures 4–23 ns when the store's nodes sit densely packed — so 50 ns is the conservative planning figure and the thresholds below follow it.) It runs on the punctuation lane, which the commit barrier waits for, so at that size it stretches every commit of the whole application. The RocksDB session store does not have this cost (it drops whole segments).

Two things make the cliff visible before it hurts: the stoatflow.store.num.keys gauge shows the live count per in-memory store, and the store logs one warning per store the first time an expiry pass takes longer than 15 ms (a tenth of the barrier floor — about 300 000 sessions on the default scan) on a store holding at least 100 000 live sessions (so a GC pause under a small store cannot trip it). Both expiry modes are timed: without the index the warning names the opt-in below; with it, the removal volume and the RocksDB store.

The opt-in is a third argument on both in-memory session factories. It adds an end-time index beside the map, consulted only by expiry, so an expiry pass costs a logarithmic lookup plus whatever actually expires, whatever the live count:

// DSL: the explicit supplier carries the switch (it wins over withStoreType, as in Kafka Streams)
stream.groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
    .count(Materialized.`as`(Stores.inMemorySessionStore("sessions", Duration.ofHours(2), expiryIndex = true)))

// Processor API
builder.addStateStore(
    Stores.sessionStoreBuilder(
        Stores.inMemorySessionStore("sessions", Duration.ofHours(2), expiryIndex = true),
        Serdes.String(), Serdes.Long(),
    ),
)
// The headers-aware twin takes the same argument: Stores.inMemorySessionStoreWithHeaders(name, retention, true)

// DSL: the explicit supplier carries the switch (it wins over withStoreType, as in Kafka Streams)
stream.groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
    .count(Materialized.as(Stores.inMemorySessionStore("sessions", Duration.ofHours(2), true)));

// Processor API
builder.addStateStore(
    Stores.sessionStoreBuilder(
        Stores.inMemorySessionStore("sessions", Duration.ofHours(2), true),
        Serdes.String(), Serdes.Long()));
// The headers-aware twin takes the same argument: Stores.inMemorySessionStoreWithHeaders(name, retention, true)

What it trades. The index moves the cost of expiry from how many sessions are alive to how many expire. The per-pass walk over the whole store goes away — about a microsecond at 500 000 sessions instead of milliseconds, and flat as the store grows. In exchange, every write pays one extra index insert (about 0.3 µs), every live session costs about 60 B more heap, and each session that does expire costs more to remove (roughly four times as much, because the index removes sessions in end-time order rather than key order). In a steady state where sessions live longer than a minute or two, the walk it removes dwarfs what it adds. Where it loses is a pass that expires a large fraction of the store at once — a watermark jump after a restore or catch-up, or sessions that end in bursts — which costs more with the index than without.

Decide on your workload, not on these numbers. They come from one host and one synthetic shape. Before opting in, run your topology under realistic traffic with the index off and on, and compare what matters to you: commit latency (stoatflow.barrier.commit.latency, since the expiry pass runs on the lane the barrier waits for), the live count (stoatflow.store.num.keys), heap, and whether the once-per-store expiry warning fires. The repository's InMemorySessionStoreExpiryCostMeasurementTest is the same comparison at store level if you want a number without a broker. As a starting point:

• Below ~100 000 live sessions the default scan costs a few milliseconds per watermark advance at most — leave the index off.
• From ~100 000 live sessions per store (the warning fires at ~300 000) the index is usually the right choice — confirm it against your session churn.
• Past a few million live sessions, or when most of the store expiring in one pass is your steady state rather than an exception, use persistentSessionStore: heap is the limit in the first case, and dropping whole segments beats removing sessions one by one in the second.

Nothing about the index is persisted, so enabling it on an existing application is an ordinary restart: the changelog restore that fills the store fills the index. Under the TopologyTestDriver, a RocksDB session store coerced to its in-memory twin runs without the index (a persistent supplier has no such switch) — that affects tests only. This is a StoatFlow extension; Kafka Streams' in-memory session store is laid out end-time-first instead and pays for cheap expiry on every single-key read.

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

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

stoatflow:
  dsl:
    store-format: HEADERS   # table, KTable-operator, join-result and aggregation stores persist 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:

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

Pick the factory by the value shape you want back. Every DSL materialisation — table(), toTable(), globalTable(), a KTable operator (filter, mapValues, transformValues, joins), a grouped or cogroup aggregation and every windowed aggregation — builds a timestamped store, and withRecordHeaders() a headers-aware one, as in Kafka Streams. The factories convert between them:

• keyValueStore() and windowStore() accept a timestamped store or its headers-aware twin and answer with the plain value, so count() reads back as Long.
• timestampedKeyValueStore() and timestampedWindowStore() accept the headers-aware twin too, and answer with ValueAndTimestamp: the timestamp without the headers.
• sessionStore() accepts a headers-aware session store and answers with the plain aggregate.

Use a *WithHeaders() factory to read the headers themselves.

Every view is read-only, as in Kafka Streams, in one of two ways. When the store holds the value shape the factory promises, the view keeps the store's own interface and a write through it (put, delete, remove, expireSessions) throws UnsupportedOperationException. That holds for the two headers-aware session and versioned types too, whose factories return the full interface with its write methods. For a headers-aware session store you can hold the result as ReadOnlySessionStore<K, AggregationWithHeaders<AGG>> instead, which SessionStoreWithHeaders extends; the versioned headers-aware type has no read-only interface. When the factory converts (keyValueStore() over a timestamped store, and the other pairs above), you get a converting view that implements only the ReadOnly* interface: it has no write methods at all, and casting it to a writable store type fails with ClassCastException. Program against the ReadOnly* type you asked for. State is written by the processors connected to the store, inside the topology, where each write is part of a commit and reaches the changelog.

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.

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

A query reads the store's latest state, including writes that are not committed yet. Reading as of a commit is what processing lanes do under exactly_once (joins under exactly-once); a query is not a processing lane.

store(...) throws IllegalStateException unless the application is RUNNING (or RESTORING / VALIDATING_STATE when enableStaleStores() was set), UnknownStateStoreException if the named store does not exist, and InvalidStateStoreException (as in Kafka Streams) if the store's type does not match the requested QueryableStoreType — a built-in type checks at the call, a custom type on its first read. storeNames() lists every registered store name for discovery.

A store handle looks the store up again on every read, so you can keep one for the life of the application. An in-place engine restart (for example a REPLACE_THREAD recovery) closes every store and opens it again; while that is under way a read through the handle throws InvalidStateStoreException, and once the application is RUNNING again the same handle reads the reopened store. After an HA demote the instance becomes a standby: it reports RUNNING, but a read through the handle keeps throwing InvalidStateStoreException until this instance is promoted again, so send queries to the new active instance. Retry on InvalidStateStoreException, as you would around a rebalance in Kafka Streams. An iterator opened before a restart is closed with its store. A handle also keeps serving while the application is paused (pause() stops processing, not the stores), although a new store(...) call needs RUNNING. With enableStaleStores(), a store(...) call made while the stores are still being restored throws InvalidStateStoreException; retry it.

A custom QueryableStoreType implements Kafka Streams' create(StateStoreProvider storeProvider, String storeName). Call storeProvider.stores(storeName, this) on every read rather than keeping the store it returns, so your view survives a restart too. For a read-only view of a built-in store family, delegate to the matching built-in type: QueryableStoreTypes.keyValueStore().create(storeProvider, storeName) — it also converts a timestamped store's values to the plain value, as the built-in type does. To hand out the store exactly as it is instead, wrap the store stores(...) returned in a public ReadOnlyKeyValueStoreWrapper(store, message) (or the window / session one): it converts nothing, and it stays bound to that instance, so it does not survive a restart.

Key metadata

queryMetadataForKey answers "where does this key live?" with Kafka Streams' KeyQueryMetadata. StoatFlow is a single instance, so activeHost() is always this instance (its application.server) and standbyHosts() is empty; partition() is the key's partition in the store's changelog topic (or its source topic, for a store restored from one), and -1 for a store no topic backs — logging disabled and no source topic — since its data lives only in the local store. It is informational: every store is global, so StoatFlow never routes by it. There are four overloads, as in Kafka Streams 4.4:

If your serializer or partitioner uses headers, pass them, or the reported partition may not be the one the key was written to. The partitioner is called with a null value, as in Kafka Streams, so in Kotlin declare its value type nullable (StreamPartitioner<String, Value?>) if you use it here. A partitioner must return a single partition here: an empty or multi-partition set throws IllegalArgumentException, as in Kafka Streams, and so does Optional.empty() (Kafka Streams returns metadata without a usable partition instead). An unknown store throws UnknownStateStoreException.

store(...), storeNames() and queryMetadataForKey(...) 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). Reads see the latest writes, cached or not. commitBarrier() commits the open epoch; with the default per-pipe commit every pipeInput already does, so call it yourself only with test.driver.commit-per-pipe=false:

import io.stoatflow.testutils.TopologyTestDriver

val driver = TopologyTestDriver.fromBuilder(builder)
// ... pipe input ...
driver.commitBarrier() // commit the epoch (only needed with commit-per-pipe=false)

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(); // commit the epoch (only needed with commit-per-pipe=false)

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

Related

• 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