A state store is Kafka Streams’ local key-value storage for stateful processing. DSL aggregations, reductions, joins, and KTables create stores automatically; persistent stores are the default. Stores hold per-key results while records flow through a topology, allowing processing to make decisions from prior records instead of treating every event independently. section 7.4 section 7.5 section 8.2
Stores are backed by Kafka changelog topics. Writes are recorded in the store and its changelog, so persistent stores retain local disk contents across restarts and in-memory stores can be restored after restart. For cached aggregations, changelog records are produced when the cache flushes at commit time or when full; with caching disabled, each record reaches the store and changelog. section 7.4
Figure 7.17 As the key-value records get written to the store, they also get written to the changelog topic for data durability.
Follow the paired write: each key-value update enters the local store and its Kafka changelog, which supplies durability and restoration data. section 7.4
Treat store ownership, placement, and naming as deployment concerns:
- A stateful task exclusively accesses its assigned store. Tasks share neither data nor a store, avoiding locking and concurrent access for that store. section 7.4.3 section 10.2.4
- Kafka Streams creates store directories under STATE_DIR_CONFIG, then application ID, task ID, RocksDB implementation, and processor or supplied store name. Each instance sharing a filesystem needs a unique state directory. section 7.4.4
- Name stores explicitly with Materialized.as(...). Generated names can shift when upstream operators change, creating a new empty store and changelog rather than restoring the former state. section 7.4.5 section 7.5
- Provide Serdes for store keys and values where required, since stores work with byte arrays; a named Materialized store can also be made available to interactive queries. section 8.2 section 13.2
Name a counting store before evolving the topology
A minimal count creates a generated aggregation store. Make that identity stable by supplying Materialized.as("counting-store") to count(). If a filter is later inserted before count(), generated processor and store names may change; the explicit name prevents that name-shift failure mode. section 7.4.4 section 7.4.5
Use a store in a custom stock processor
For stock-trend analysis, create an in-memory key-value store through Stores, provide its StoreBuilder from ProcessorSupplier.stores(), then retrieve it in init() with ProcessorContext.getStateStore(). A scheduled punctuator iterates the store every 10 seconds of stream time and forwards only entries whose price or share differential reaches the threshold. section 10.2 section 10.2.1 section 10.2.4
A named store also supports direct inspection. Interactive Queries accesses local stateful results, while application instances divide stores by task and partition across machines. In tests, TopologyTestDriver can retrieve a KeyValueStore by name and assert that its contents match the latest output. section 13.2 section 13.3 section 14.1.4
