How to build a custom Kafka Streams Statestores

What is a Kafka Streams state store?

Stateful Kafka Streams operations, such as counts, aggregations, joins and de-duplication, need somewhere to remember what they have already seen. That somewhere is a state store. Each stream task gets its own store instance for the input partitions it owns, so state is sharded in the same way as your topics.

You rarely create a store directly. Instead you define a StoreBuilder, and Kafka Streams uses it as a factory to create store instances wherever tasks run. The DSL does this for you behind count(), aggregate() and toTable(); the Processor API lets you attach stores to your own processors. If you are new to the library, start with our Kafka Streams 101 guide.

State stores can also be read from outside the topology through interactive queries. Those queries are read-only and only see the local shard, so exposing the full state of an application means adding your own RPC layer, such as a REST endpoint, between instances, as the Kafka Streams interactive queries guide explains.

Which built-in state store types does Kafka Streams offer?

Kafka Streams ships two families of store, both described in the Processor API documentation:

Persistent stores are the default, and RocksDB can be tuned through rocksdb.config.setter. In-memory stores avoid disk entirely but must be rebuilt from the changelog on every restart, and there is no built-in in-memory versioned store. Whatever you choose, always close iterators from a store, ideally with try-with-resources, because an unclosed iterator can cause out-of-memory errors.

How do changelogs and standby replicas keep state safe?

Every fault-tolerant store is continuously backed up to a compacted changelog topic. If an instance fails, or a task moves to another machine during a rebalance, Kafka Streams replays the changelog to rebuild the store before it resumes processing. Window store changelogs use compaction and deletion together so that expired windows are cleaned up.

Replaying a large changelog can take a long time, which is where standby replicas help. A standby is a fully replicated copy of a task’s state kept warm on another instance:

According to the Kafka Streams configuration reference, the default state.dir is not a good fit for production, and the docs’ own resilience example sets one standby replica. If you disable change logging on a store with withLoggingDisabled(), that store is no longer fault tolerant and cannot have standby replicas.

Key takeaway: Treat the changelog as the source of truth and the local store as a fast, rebuildable cache. Give it a persistent state.dir and at least one standby replica, and most restore-time problems disappear.

When should you build a custom state store?

Most teams never need one. RocksDB copes with large state, and the DSL covers the common aggregations. A custom store makes sense when your processor needs something the built-in stores cannot do efficiently, for example:

• Secondary indexes or rich queries, such as looking up records by a field other than the key, where a key-value store would force a full scan.
• A different storage engine, such as an embedded document store like Nitrite, H2 or Lucene, when you want queries pushed down into the engine.
• Special retention or compaction rules that windowed and session stores do not express.

Keep the store embedded and owned by its task. A tempting shortcut is to point several application instances at one shared external database. That breaks the one-task, one-shard model Kafka Streams relies on, makes restores and rebalances unsafe, and weakens the processing guarantees. If other systems need the data, write results to an output topic and load them with a sink connector instead.

How do you implement and register a custom state store?

The primary interface is org.apache.kafka.streams.processor.StateStore, with extended interfaces such as KeyValueStore and VersionedKeyValueStore. Your implementation must also supply restore logic through StateRestoreCallback or BatchingStateRestoreCallback, so Kafka Streams can rebuild it from the changelog, and a StoreBuilder that creates instances.

For a key-value style store, the simplest route is to implement KeyValueBytesStoreSupplier around your engine and hand it to the standard builder, which adds serdes, caching and change logging for you:

StoreBuilder<KeyValueStore<String, Long>> countStoreBuilder =
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("Counts"), // or your own KeyValueBytesStoreSupplier
        Serdes.String(),
        Serdes.Long())
    .withLoggingEnabled(changelogConfig);

topology.addStateStore(countStoreBuilder, "Process");

On Kubernetes, run stateful Kafka Streams applications as a StatefulSet with a persistent volume claim mounted at state.dir. A restarted pod then finds its store on disk and only needs to catch up from the changelog rather than rebuild everything. Our guides to running Kafka consumers on Kubernetes and Kafka Streams memory best practices cover sizing, and for a different approach that moves state out of the application entirely, read how unbundled state stores change Kafka Streams operations.

If your team needs help designing stateful Kafka Streams applications or fixing slow restores in production, our Kafka support and Kafka consulting services engineers do this every day.

Frequently asked questions