State Store #

When designing real-time data processing microservices architectures, one of the most complex problems we often face is how to manage state storage during stateful computations. Operations like hourly revenue aggregation, user click frequency counting, or joining two event streams heavily depend on the system’s ability to store and read state quickly. Apache Kafka provides a storage abstraction called the State Store to solve this problem. Kafka Streams’ State Store combines local memory access speed through the RocksDB engine with distributed data resilience. Through this article, we’ll deeply dissect the internal State Store architecture, the write-read paths involving the JVM Heap Cache, RocksDB, and Changelog Topics, manual RocksDB performance tuning techniques, and how to leverage Interactive Queries (IQ) to safely expose local state data to the outside world without triggering latency degradation.

The Important Role of Local State Stores in Stateful Processing #

When doing traditional stream data processing, centralized external databases (like Redis, PostgreSQL, or Cassandra) are often used as state storage containers. However, this approach has enormous scalability limitations at high throughput:

  • Network Latency Overhead: Every processed data record must do round-trip network calls (RPC) to the external database. If we process 50,000 records per second, this network load triggers severe bottlenecks.
  • ACID Challenges: Managing distributed transactional data consistency across systems (Kafka and external databases) is very difficult, especially if network failures happen mid-processing.

The Kafka Solution: Local State Stores #

To eliminate these network latency obstacles, Kafka Streams applies the principle of unifying computation with data. Every application instance stores its state data chunks locally in an embedded storage engine (Local State Store).

  • Zero-Network Access: When a StreamTask filters or updates a running account balance, it writes the new balance to the local disk or RAM memory of the server machine where the application itself runs. The latency is microseconds, not milliseconds.
  • Task Isolation: Every State Store strictly only loads data for the input topic partitions assigned to that StreamTask. There’s no locking contention between different processing threads.

Why RocksDB? Stateful Processing Outside the JVM Heap #

By default, Kafka Streams uses RocksDB as the local persistent storage engine for State Stores. RocksDB is an embedded key-value store built by Meta (Facebook), written in C++, highly optimized for fast SSD storage media.

Preventing Long GC Pauses (Off-Heap Memory) #

One of the main reasons RocksDB is chosen, instead of a regular Java HashMap inside JVM memory, is to avoid the severe impact of Garbage Collection (GC) Pauses.

  • The JVM Heap Problem: If we store millions of user profile data inside JVM Heap memory using regular Java objects, the JVM Garbage Collector must scan those millions of objects every cleanup cycle. This stops our application’s entire thread execution (Stop-the-World GC Pauses) for seconds to minutes, destroying real-time latency guarantees.
  • The Off-Heap RocksDB Solution: RocksDB stores its data outside JVM Heap memory (Off-Heap Memory). RocksDB operates directly in the operating system kernel memory space and uses LSM-Tree (Log-Structured Merge-tree) techniques to write data to disk sequentially. Data is only converted from binary to Java objects momentarily when needed by application logic, keeping the JVM Heap clean and lightweight.

Data Write Synergy: State Store Write and Read Flows #

To balance between memory read-write speed and physical data resilience from power failures, Kafka Streams implements a three-layer integrated processing flow:

  1. JVM Heap Cache (Speed Layer) Every State Store has an internal memory cache inside the JVM Heap (configurable via cache.max.bytes.buffering). When new records arrive, data is written to this cache first for fast in-memory deduplication.
  2. RocksDB Storage Engine (Local Persistent Layer) Periodically (or when the JVM memory cache is full), data is flushed to the local RocksDB database instance stored on the server SSD.
  3. Changelog Topic (Distributed Durability Layer) At the same time data is written to RocksDB, those changes are sent as changelog events to an internal topic on the Kafka broker (application-id-store-name-changelog). This topic acts as life insurance for our local state data.
flowchart TD
    subgraph APP["Java KStreams Application (StreamTask)"]
        direction TB
        Logika["Business Logic (e.g. aggregate)"]
        JVMCache["JVM Heap Cache (Memory Buffering)"]
    end

    subgraph LOCAL["Local Physical Storage (Host SSD)"]
        RocksDB[("RocksDB Store (Off-Heap C++)")]
    end

    subgraph BROKER["Kafka Broker Infrastructure"]
        Changelog["Topic: application-id-user-store-changelog"]
    end

    Logika -->|"1. Fast write"| JVMCache
    JVMCache -->|"2. Periodic flush (Off-Heap)"| RocksDB
    JVMCache -.->|"3. Async Replication (Record Update)"| Changelog

    style Logika stroke:#0288d1,stroke-width:2px
    style JVMCache stroke:#388e3c,stroke-width:2px
    style RocksDB stroke:#f57c00,stroke-width:2px
    style Changelog stroke:#d32f2f,stroke-width:2px

RocksDB Customization and Performance Tuning #

Although RocksDB is very robust, Kafka Streams’ default configuration is sometimes not optimal for all computation workload types. For super-high write volume scenarios, the built-in RocksDB can experience slowdowns from memory block division (compaction style). We can adjust RocksDB using the RocksDBConfigSetter implementation.

Here’s an example of customizing RocksDB memory parameters to minimize disk flush pause times and enlarge block cache capacity:

// CORRECT: Implementing RocksDBConfigSetter to optimize RocksDB off-heap memory allocation.
// ✓ Controlling block caches, write buffer sizes, and avoiding disk write stalls.
public class CustomRocksDbConfig implements RocksDBConfigSetter {
    @Override
    public void setConfig(String storeName, Options options, Map<String, Object> configs) {
        // 1. Create a block filter policy to speed up data lookups with Bloom Filters
        org.rocksdb.BlockBasedTableConfig tableConfig = new org.rocksdb.BlockBasedTableConfig();
        
        // Set the block cache size (e.g., 64MB)
        tableConfig.setBlockCache(new org.rocksdb.LRUCache(64 * 1024 * 1024L));
        // Set the data block size (default 4KB)
        tableConfig.setBlockSize(4 * 1024L);
        // Add a bloom filter with 10 bits per key (reducing false disk reads)
        tableConfig.setFilterPolicy(new org.rocksdb.BloomFilter(10, false));
        
        options.setTableFormatConfig(tableConfig);

        // 2. Optimize the Write Buffer (MemTable) size
        // writeBufferSize determines the memory limit per partition before flushing to disk (e.g., 16MB)
        options.setWriteBufferSize(16 * 1024 * 1024L);
        // The maximum number of write buffers piling up in memory before being blocked
        options.setMaxWriteBufferNumber(3);

        // 3. Configure data compression to save local SSD storage space
        options.setCompressionType(CompressionType.LZ4_COMPRESSION);
    }

    @Override
    public void close(String storeName, Options options) {
        // Clean up resources if any
    }
}

To register the custom configuration above into our Kafka Streams application, we just add it to the configuration properties:

Properties props = new Properties();
props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, CustomRocksDbConfig.class.getName());

Implementation Code: Anti-Pattern vs the Official State Store Solution #

Let’s compare the dangerous implementations often encountered with the correct way using managed State Stores in Kafka Streams.

Use Case Scenario #

We want to track consecutive failed login counts for e-commerce users (failed-login-events). If a user fails to log in more than 3 consecutive times, we want to flag that user for additional verification.

Anti-Pattern: Using a Simple In-Memory HashMap Without an Official State Store #

// ANTI-PATTERN: Using an internal Java memory Map inside stream processing.
// ✗ Prone to losing all data on crashes, not supported by Changelog backups, and triggers OutOfMemory.
public class VolatileMemoryProcessor {
    // ✗ DANGER: This Map disappears on restart and isn't distributed across tasks
    private static final Map<String, Integer> failedLoginCounts = new ConcurrentHashMap<>();

    public static void processStream(KStream<String, String> stream) {
        stream.foreach((userId, eventJson) -> {
            int currentCount = failedLoginCounts.getOrDefault(userId, 0) + 1;
            failedLoginCounts.put(userId, currentCount);

            if (currentCount >= 3) {
                log.warn("User {} has exceeded 3 failed login attempts!", userId);
            }
        });
    }
}

Practical Solution: Defining a Managed Persistent State Store #

Below is the correct implementation by registering a crash-safe RocksDB-based persistent State Store.

// CORRECT: Using KStream.process() to access persistent RocksDB State Stores.
// ✓ Safe from crashes, connected to Changelog Topics for automatic status replication.
public class SecureStateStoreProcessor implements Processor<String, String, String, String> {
    private KeyValueStore<String, Integer> failedLoginStore;
    private ProcessorContext<String, String> context;

    @Override
    public void init(ProcessorContext<String, String> context) {
        this.context = context;
        // ✓ Getting a reference to the registered RocksDB store
        this.failedLoginStore = context.getStateStore("failed-login-store");
    }

    @Override
    public void process(Record<String, String> record) {
        String userId = record.key();
        
        // Safely reading the current failed login count from local RocksDB
        Integer currentCount = failedLoginStore.get(userId);
        if (currentCount == null) {
            currentCount = 0;
        }

        int newCount = currentCount + 1;
        
        // Store the new value to local RocksDB
        failedLoginStore.put(userId, newCount);

        if (newCount >= 3) {
            // Forward the suspicious transaction alert downstream
            context.forward(new Record<>(userId, "BLOCKED", record.timestamp()));
        }
    }

    @Override
    public void close() { }
}

Interactive Queries (IQ): Accessing State Directly #

One of the most revolutionary features in Kafka Streams is Interactive Queries (IQ). Usually, when applications write processed result data to databases, other applications must query that external database. With IQ, State Stores inside the Kafka Streams application can be queried directly by external services through REST APIs without first writing data to an external database.

However, because our KTable data is locally divided across several partitions and instances, we must do a two-stage lookup:

  1. Determine which instance holds the partition for the key being searched.
  2. Make an HTTP REST call to that instance if the key isn’t in the current instance’s local memory.

Implementation Code: Querying State Stores via Interactive Queries #

Here’s a Java code example for finding instance metadata locations and retrieving data from local State Stores.

// CORRECT: Implementing Interactive Queries to expose local State Store data.
// ✓ Using metadata to determine the key partition location before reading the query.
public class InteractiveQueryService {
    private final KafkaStreams streams;

    public InteractiveQueryService(KafkaStreams streams) {
        this.streams = streams;
    }

    public Integer getFailedLoginCount(String userId) {
        // First connect the streams liveness check
        if (streams.state() != KafkaStreams.State.RUNNING) {
            throw new IllegalStateException("The Kafka Streams cluster is in rebalance or idle status.");
        }

        // Step 1: Find out which broker metadata holds that key's partition
        KeyQueryMetadata metadata = streams.queryMetadataForKey(
            "failed-login-store",
            userId,
            Serdes.String().serializer()
        );

        if (metadata == null) {
            throw new RuntimeException("Metadata not found for key: " + userId);
        }

        // Step 2: Evaluate whether that key's partition exists on the current active instance
        if (metadata.activeHost().port() == getCurrentInstancePort()) {
            // ✓ The key is in local memory, read directly from local RocksDB without network
            ReadOnlyKeyValueStore<String, Integer> store = streams.store(
                StoreQueryParameters.fromNameAndType(
                    "failed-login-store",
                    QueryableStoreTypes.readOnlyKeyValueStore()
                )
            );
            return store.get(userId);
        } else {
            // ✓ The key is on another instance, direct the HTTP Request to the destination host
            String targetUrl = String.format("http://%s:%d/users/%s/failed-logins",
                metadata.activeHost().host(),
                metadata.activeHost().port(),
                userId
            );
            return fetchFromRemoteInstance(targetUrl);
        }
    }

    private int getCurrentInstancePort() { return 8080; }
    private Integer fetchFromRemoteInstance(String url) { return 0; /* Simulated REST call */ }
}

Summary #

  • Local State Store — The integrated local state data storage interface uniting data with computation to avoid external network call latency.
  • RocksDB Engine — A key-value database embedded with LSM-Tree architecture operating outside JVM heap memory (off-heap) to avoid Stop-the-world GC Pauses.
  • Write-Read Path — State changes are first written to the JVM cache, followed by flushes to local RocksDB and asynchronous backups to Changelog Topics on Kafka brokers.
  • Changelog Topic — The high-replication internal topic recording every state store data transition to guarantee local data recovery during node failures.
  • RocksDBConfigSetter — The special Java interface for customizing detailed RocksDB off-heap memory parameters like block cache size and compression type.
  • Interactive Queries (IQ) — The mechanism for directly querying application-local State Stores from external REST APIs without moving data to a separate database.
  • num.standby.replicas — The passive state store replica configuration on other instances to speed up state recovery without cold start recovery pause times.

← Previous: Topology vs Processor Next: Changelog Topic →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact