Home/Learn/Apache Kafka/State Stores & RocksDB

State Stores & RocksDB

Advanced

Kafka Streams

Stateful operations (aggregations, joins) persist state in RocksDB-backed changelog topics; state is restored automatically on restart, enabling fault-tolerant stateful processing.

Overview

Kafka Streams uses state stores to maintain local state for stateful operations like aggregations, windowed counts, and stream-table joins. By default, persistent state stores use RocksDB — an embedded key-value store that writes to local disk. Every state store is backed by a changelog Kafka topic: on every state change, a record is written to the changelog. When a stream instance restarts or a partition is reassigned, state is restored by replaying the changelog. This makes stateful Kafka Streams applications fault-tolerant without a central database. In-memory state stores are available for low-latency scenarios where recovery speed is traded for simplicity, though they lose state on restart.

Persistent state store with RocksDB

Persistent state stores (default) write to RocksDB on local disk and emit a changelog to a compacted Kafka topic. The changelog topic is created automatically with the name <application-id>-<store-name>-changelog. On restart, Kafka Streams replays the changelog to rebuild local state. Standby replicas (num.standby.replicas) pre-build state on secondary instances for faster failover.

Java — persistent state store backed by RocksDB with changelog topic

// Word count with persistent state store (default)
StreamsBuilder builder = new StreamsBuilder();

KTable<String, Long> wordCounts = builder
    .stream("sentences-input", Consumed.with(Serdes.String(), Serdes.String()))
    .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
    .groupBy((key, word) -> word)
    .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("word-count-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long()));
        // ↑ "word-count-store" is backed by RocksDB
        // changelog topic: myapp-word-count-store-changelog

wordCounts.toStream().to("word-counts-output");

// application.properties for standby replicas
// spring.kafka.streams.properties.num.standby.replicas=1
// Standby instance keeps an up-to-date copy for fast failover

In-memory state stores and custom stores

In-memory stores use Materialized.as(Stores.inMemoryKeyValueStore(name)) — faster than RocksDB for small datasets but lose state on JVM restart (must replay full changelog). Custom stores let you implement the StateStore interface for specialised storage, such as off-heap stores or stores backed by an external cache like Redis.

Java — in-memory store and direct state store querying

// In-memory state store — faster, but loses state on restart
KTable<String, Long> wordCounts = builder
    .stream("sentences-input", Consumed.with(Serdes.String(), Serdes.String()))
    .flatMapValues(value -> Arrays.asList(value.split(" ")))
    .groupBy((k, v) -> v)
    .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as(
        Stores.inMemoryKeyValueStore("word-count-inmemory")
    ).withKeySerde(Serdes.String()).withValueSerde(Serdes.Long()));

// Querying the state store directly (Interactive Queries)
ReadOnlyKeyValueStore<String, Long> store = streams
    .store(StoreQueryParameters.fromNameAndType(
        "word-count-store",
        QueryableStoreTypes.keyValueStore()));

Long count = store.get("kafka");                    // point lookup
KeyValueIterator<String, Long> all = store.all();   // scan all entries
KeyValueIterator<String, Long> range = store.range("a", "m");  // range scan

State restoration and changelog management

On startup or rebalance, Kafka Streams restores state by replaying the changelog topic from the beginning (or a recent snapshot offset). Restoration time depends on changelog topic size. To speed this up: enable standby replicas, use rocksdb.block.cache settings, or publish periodic state snapshots. The StateRestoreListener callback lets you monitor restoration progress.

Java — StateRestoreListener for monitoring changelog restoration

// Monitor state restoration progress
public class LoggingStateRestoreListener implements StateRestoreListener {

    @Override
    public void onRestoreStart(TopicPartition tp, String storeName,
                               long startOffset, long endOffset) {
        log.info("Restoring store={} partition={} messages={}",
            storeName, tp, endOffset - startOffset);
    }

    @Override
    public void onBatchRestored(TopicPartition tp, String storeName,
                                long batchEndOffset, long numRestored) {
        log.debug("Restored {} records for store={}", numRestored, storeName);
    }

    @Override
    public void onRestoreEnd(TopicPartition tp, String storeName, long totalRestored) {
        log.info("Restoration complete store={} total={}", storeName, totalRestored);
    }
}

// Register the listener
KafkaStreams streams = new KafkaStreams(topology, props);
streams.setGlobalStateRestoreListener(new LoggingStateRestoreListener());
streams.start();

// application.properties — RocksDB tuning
// spring.kafka.streams.properties.rocksdb.config.setter=com.example.RocksDbConfig

Key Points to Remember

1State stores back stateful operations (aggregations, joins) with local RocksDB storage and a Kafka changelog topic
2Changelog topics are compacted Kafka topics auto-created with name <app-id>-<store-name>-changelog
3On restart, state is restored by replaying the changelog — this is what makes Kafka Streams fault-tolerant without external DB
4In-memory stores (Stores.inMemoryKeyValueStore) are faster but lose state on restart and must replay full changelog
5Standby replicas (num.standby.replicas) pre-build state on secondary instances, reducing restoration time on failover
6Interactive Queries API (streams.store(…)) lets you query local state stores directly as a read model

Interview Questions

How does Kafka Streams achieve fault-tolerant stateful processing without a shared database?

MediumConfluent

What is the changelog topic and what happens to it when you delete a state store?

MediumLinkedIn

What are the trade-offs between persistent (RocksDB) and in-memory state stores?

MediumUber

How do standby replicas reduce state restoration time after a failover?

HardNetflix

How would you monitor and alert on slow state restoration in a production Kafka Streams application?

HardAmazon

Ask Aria about State Stores & RocksDB

Your personal AI tutor — ask anything about this concept

Revision Status

Personal Notes

Discussion

Loading discussion…

Windowing in Kafka Streams

Interactive Queries