Interactive Queries and State Stores in Kafka Streams

In stream processing architectures, querying the current state of a running stream application is an operational necessity. Stream processors accumulate valuable, precomputed state in local storage, yet accessing that state from outside the application is non-trivial. Interactive Queries in Kafka Streams close this gap: they expose the contents of state stores through a read-only API, turning each application instance into a queryable service for the partitions it owns.

Instead of sinking aggregated results to an external database solely for read access, you can query the state store directly, removing a moving part from the architecture and cutting end-to-end read latency to a local RocksDB lookup. For SRE and backend engineers running Kafka in production, the mechanics, routing model, and failure modes of Interactive Queries determine whether that local-first read path is actually reliable.

This article covers the architecture of state stores, the API for exposing them, query routing across instances, and the operational concerns—rebalancing, RocksDB tuning, readiness, security, and monitoring—that make the feature production-grade. It assumes familiarity with the Kafka Streams DSL and stateful processing; for a refresher on choosing between the DSL and the lower-level Processor API, see Kafka Streams DSL vs Processor API: When to Use Each.

The Architecture of State Stores in Kafka Streams

State stores back stateful stream processing. Any groupByKey().aggregate(), materialized KTable, or windowed operation causes Kafka Streams to materialize intermediate state in a local, embedded store. How these stores are provisioned, backed up, and restored is the prerequisite to exposing them safely.

Local Storage and Changelog Topics

Each stream task maintains its own state store, backed by RocksDB by default. The store lives on the local filesystem of the host running the stream thread. Crucially, every logged store is paired with an internal changelog topic in the cluster. Updates to the local store are also written to this changelog, providing a durable, replicated log of state mutations.

This serves two purposes. First, fault tolerance: if an instance dies, another instance rebuilds the store by replaying the changelog. Second, it underpins the metadata that drives query routing. The changelog is co-partitioned with the store’s input, so the state for a given key resides on exactly one active task at a time.

Changelog topics are named <application.id>-<store-name>-changelog. To inspect the changelog for a store named user-activity-store in an application with application.id=my-stream-app:

kafka-console-consumer.sh \
  --bootstrap-server kafka-broker-1:9092 \
  --topic my-stream-app-user-activity-store-changelog \
  --from-beginning \
  --property print.key=true \
  --key-deserializer org.apache.kafka.common.serialization.StringDeserializer \
  --value-deserializer org.apache.kafka.common.serialization.LongDeserializer

--key-deserializer and --value-deserializer are the correct console-consumer arguments; the serdes here assume a String key and Long value, which you should match to your store’s actual serdes.

State Store Types and Configuration

Kafka Streams offers persistent stores (default, RocksDB-backed) and in-memory stores. Persistent stores survive restarts and suit state larger than RAM. In-memory stores avoid RocksDB overhead but must be fully rebuilt from the changelog on restart. Either way, recovery for a logged store comes from the changelog—the difference is whether warm local data survives a restart.

Materialize a key-value store and customize the changelog topic config in the topology:

Properties props = new Properties();
props.put(StreamsConfig.STATE_DIR_CONFIG, "/data/kafka-streams-state");
props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG,
          CustomRocksDBConfig.class.getName());

Map<String, String> changelogConfig = new HashMap<>();
changelogConfig.put("min.insync.replicas", "2");

Materialized<String, Long, KeyValueStore<Bytes, byte[]>> materialized =
    Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("user-activity-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long())
        .withLoggingEnabled(changelogConfig);

STATE_DIR_CONFIG controls where RocksDB files are written; in containerized environments mount a persistent volume there so a pod restart restarts from warm local state rather than a full changelog replay. withLoggingEnabled(Map) passes topic-level configs (here min.insync.replicas) to the changelog topic; unrecognized keys are ignored.

Exposing State Stores with Interactive Queries

Interactive Queries let external clients issue read-only queries against the state stores of a running application. The query API is embedded in each instance, so an instance can serve queries for the partitions it actively hosts.

The Query API and Store Access

The entry point is KafkaStreams.store(StoreQueryParameters). You pass the store name and a queryable store type and get back a read-only handle for key lookups or range scans. A REST endpoint over a key-value store:

@GET
@Path("/state/{key}")
public Response queryState(@PathParam("key") String key) {
    ReadOnlyKeyValueStore<String, Long> store =
        streams.store(StoreQueryParameters.fromNameAndType(
            "user-activity-store",
            QueryableStoreTypes.keyValueStore()));
    Long value = store.get(key);
    return value != null
        ? Response.ok(value).build()
        : Response.status(404).entity("Key not found").build();
}

The lookup is local—no hop to an external database—but it only returns data if the key’s partition is hosted by this instance. For correct results across a cluster you must route.

Note that store() throws InvalidStateStoreException while the instance is rebalancing or restoring; production code should catch it and retry rather than 500.

Query Routing Across Instances

Each partition is owned by exactly one active task, and a key’s partition is its serialized-key hash modulo the partition count. To find the instance hosting a key, call KafkaStreams.queryMetadataForKey(storeName, key, keySerializer), which returns a KeyQueryMetadata. Its activeHost() returns a HostInfo—use host() and port() on that, not a (non-existent) activePort() method. The method returns KeyQueryMetadata.NOT_AVAILABLE when metadata is temporarily unknown (for example, mid-rebalance), and null if no matching metadata exists at all.

The standard pattern embeds an HTTP server (with the same host:port advertised via application.server) in each instance and forwards requests to the owning host:

flowchart TD A["Client request for key"] --> B["queryMetadataForKey"] B --> C{"activeHost is this instance?"} C -- "Yes" --> D["Local RocksDB lookup"] C -- "No" --> E["Forward to owning host over RPC"] E --> F["Remote RocksDB lookup"] D --> G["Return value"] F --> G
@GET
@Path("/state/{key}")
public Response queryOrRedirect(@PathParam("key") String key) {
    KeyQueryMetadata metadata = streams.queryMetadataForKey(
        "user-activity-store", key, Serdes.String().serializer());

    if (metadata == null || metadata.equals(KeyQueryMetadata.NOT_AVAILABLE)) {
        return Response.status(503).entity("Metadata not available; retry").build();
    }

    HostInfo activeHost = metadata.activeHost();
    if (activeHost.host().equals(localHost) && activeHost.port() == localPort) {
        try {
            ReadOnlyKeyValueStore<String, Long> store =
                streams.store(StoreQueryParameters.fromNameAndType(
                    "user-activity-store",
                    QueryableStoreTypes.keyValueStore()));
            Long value = store.get(key);
            return value != null ? Response.ok(value).build()
                                 : Response.status(404).build();
        } catch (InvalidStateStoreException e) {
            return Response.status(503).entity("Store rebalancing; retry").build();
        }
    }

    URI redirectUri = URI.create(String.format(
        "http://%s:%d/state/%s", activeHost.host(), activeHost.port(), key));
    return Response.temporaryRedirect(redirectUri).build();
}

For the comparison against localHost/localPort, use the exact application.server value you configured for this instance rather than the OS hostname, so the equality check matches what the metadata reports. In production you typically front the instances with a thin proxy (or a service mesh) so callers do not have to follow 307 redirects themselves. Confluent documents this REST pattern in its Interactive Queries developer guide.

To tolerate active-instance downtime, you can also read from standby replicas by inspecting metadata.standbyHosts() and routing there when the active host is unreachable; standby reads may be slightly stale.

Querying Windowed and Session Stores

Windowed stores use compound keys (record key plus window start), so you query a time range. fetch(key, Instant, Instant) returns a WindowStoreIterator<V> whose entries have a Long key (the window start timestamp) and the stored value:

ReadOnlyWindowStore<String, Long> windowStore =
    streams.store(StoreQueryParameters.fromNameAndType(
        "user-activity-windowed-store",
        QueryableStoreTypes.windowStore()));

Instant from = Instant.now().minus(Duration.ofHours(1));
Instant to = Instant.now();

try (WindowStoreIterator<Long> iterator = windowStore.fetch("user-123", from, to)) {
    while (iterator.hasNext()) {
        KeyValue<Long, Long> entry = iterator.next();
        System.out.println("Window start: " + entry.key + ", value: " + entry.value);
    }
}

Session stores are queried with QueryableStoreTypes.sessionStore(). Always close iterators (a try-with-resources block does this) to release the underlying RocksDB iterator and snapshot.

Operational Considerations for Production

Interactive Queries inherit every operational concern of a stateful Streams app: store sizing, rebalancing, and recovery. These directly determine query availability and tail latency.

State Store Sizing and RocksDB Tuning

RocksDB performance is sensitive to configuration, and the defaults are conservative. Tune it through a RocksDBConfigSetter. The block cache is org.rocksdb.LRUCache—not any Guava cache—and the filter policy and write buffers come from the org.rocksdb package:

public class CustomRocksDBConfig implements RocksDBConfigSetter {

    // One cache shared across all stores in the JVM keeps total off-heap
    // memory bounded; a per-store cache would multiply by the store count.
    private static final org.rocksdb.Cache CACHE =
        new org.rocksdb.LRUCache(512L * 1024 * 1024);  // 512 MB block cache

    @Override
    public void setConfig(String storeName, Options options, Map<String, Object> configs) {
        BlockBasedTableConfig tableConfig =
            (BlockBasedTableConfig) options.tableFormatConfig();
        tableConfig.setBlockCache(CACHE);
        tableConfig.setBlockSize(16L * 1024);                    // 16 KB blocks
        tableConfig.setCacheIndexAndFilterBlocks(true);
        tableConfig.setFilterPolicy(new BloomFilter(10, false)); // 10 bits/key

        options.setTableFormatConfig(tableConfig);
        options.setMaxWriteBufferNumber(3);
        options.setWriteBufferSize(64L * 1024 * 1024);           // 64 MB memtable
        options.setCompactionStyle(CompactionStyle.LEVEL);
        options.setLevelCompactionDynamicLevelBytes(true);
    }

    @Override
    public void close(String storeName, Options options) {
        // Do NOT close the shared CACHE here; it outlives individual stores.
    }
}

Two production notes. First, retrieve the existing BlockBasedTableConfig via options.tableFormatConfig() and mutate it, rather than constructing a fresh one that drops the defaults Streams already set. Second, share a single LRUCache (and, if you go further, a WriteBufferManager) across all stores so total off-heap memory is bounded regardless of how many partitions land on the instance—this is the approach in the official memory management guide.

Rebalancing and State Restoration

When instances join or leave the group, Kafka Streams rebalances and reassigns tasks. A newly assigned active task must restore its store from the changelog before it can serve queries; restoration time scales with changelog size and can range from seconds to hours.

To shrink that window, configure standby replicas. Standby tasks keep a passive copy of the store warm by tailing the changelog, so a promoted standby is queryable with little or no restore delay:

props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);

Each standby is a full extra copy of the state, consuming disk and memory per replica—trading infrastructure cost for failover speed. Combined with cooperative rebalancing (the default StreamsPartitionAssignor, which does incremental cooperative rebalancing) and a non-zero acceptable.recovery.lag, this keeps most partitions queryable across a rebalance.

Health Checks and Readiness Probes

In Kubernetes, an instance must signal readiness only when it can actually serve queries. After a restart or rebalance, the readiness endpoint should fail until stores are caught up. Inspect KafkaStreams.state() and the per-partition lag from allLocalStorePartitionLags(), which returns a nested Map<String, Map<Integer, LagInfo>> (store name → partition → lag), so you must flatten it and read LagInfo.offsetLag():

@GET
@Path("/ready")
public Response ready() {
    if (streams.state() != KafkaStreams.State.RUNNING) {
        return Response.status(503).entity("Not running").build();
    }
    boolean caughtUp = streams.allLocalStorePartitionLags().values().stream()
        .flatMap(partitionLags -> partitionLags.values().stream())
        .allMatch(lagInfo -> lagInfo.offsetLag() == 0L);

    return caughtUp ? Response.ok().build()
                    : Response.status(503).entity("Restoring").build();
}

Wire this to a Kubernetes readiness probe so traffic is withheld from instances that cannot yet answer. Use a separate liveness probe that only checks the process is up and state() is not ERROR/NOT_RUNNING, so a long but healthy restore does not trigger a restart loop.

Integrating Interactive Queries into Stream Processing Automation

Stateful Stream Processing with Kafka Streams is the foundation; Interactive Queries make those processors first-class read services in the operational data plane.

Real-Time Dashboards Without External Databases

Sinking aggregates to Redis or PostgreSQL only to power a dashboard adds a sink connector, a database, and replication lag. With Interactive Queries the dashboard backend reads the stream application directly and always reflects the latest computed state.

A global query—“top N users by activity”—requires fanning out to every instance, collecting partials, and merging, a scatter-gather. Use streamsMetadataForStore() (the non-deprecated replacement for allMetadataForStore(), which was deprecated in 3.0):

public List<KeyValue<String, Long>> topN(int n) {
    Collection<StreamsMetadata> allMetadata =
        streams.streamsMetadataForStore("user-activity-store");

    List<CompletableFuture<List<KeyValue<String, Long>>>> futures = allMetadata.stream()
        .map(meta -> CompletableFuture.supplyAsync(() ->
            isLocalInstance(meta) ? localTopN(n)
                                  : remoteTopN(meta.host(), meta.port(), n)))
        .collect(Collectors.toList());

    return futures.stream()
        .map(CompletableFuture::join)
        .flatMap(List::stream)
        .sorted((a, b) -> Long.compare(b.value, a.value))
        .limit(n)
        .collect(Collectors.toList());
}

StreamsMetadata exposes host() and port() convenience methods (equivalent to hostInfo().host() / .port()). Latency is bounded by the slowest instance, so set aggressive per-call timeouts on join() and add circuit breakers for the remote calls.

Enriching Inbound Streams with Stored State

The same store handles can enrich events inside the JVM. A fraud-detection pipeline can look up a user-profile store to attach a risk score to each transaction before scoring. For DSL-level joins of a stream against table state, prefer a KStream-KTable join, which is partition-local and avoids manual lookups; reach for in-JVM store access from the Processor API when you need imperative control. The official Interactive Queries documentation covers both store-access styles.

Security and Access Control

Exposing state over HTTP is a new attack surface. Production deployments need authentication, authorization, and encryption on the query endpoints.

Securing the Query API

Require mutual TLS between query clients and instances; on Kubernetes a mesh such as Istio can enforce mTLS transparently. For application-level authorization, validate an OAuth2/JWT token and check a scope before serving:

@GET
@Path("/state/{key}")
public Response queryState(@PathParam("key") String key,
                          @HeaderParam("Authorization") String authHeader) {
    if (!isAuthorized(authHeader, "state:read")) {
        return Response.status(403).build();
    }
    // proceed with the routed lookup
    return queryOrRedirect(key);
}

Expose only stores explicitly designed for external consumption—not every internal store should be reachable—and avoid leaking data through unbounded range scans.

Network Segmentation and Rate Limiting

Place instances in a private subnet and expose the query API only to authorized services. Front it with a reverse proxy (Envoy or NGINX) for rate limiting so one client cannot saturate RocksDB with expensive range scans:

limit_req_zone $binary_remote_addr zone=querylimit:10m rate=100r/s;

server {
    location /state/ {
        limit_req zone=querylimit burst=50 nodelay;
        proxy_pass http://kafka-streams-backend;
    }
}

Monitoring and Observability

Operating Interactive Queries needs visibility into restore progress, store health, and resource use. Kafka Streams exposes metrics via JMX; export them to your monitoring stack.

Key Metrics to Track

RocksDB store metrics are published under the stream-state-metrics MBean (added by KIP-471) at recording level DEBUG, so set metrics.recording.level=DEBUG to collect them:

kafka.streams:type=stream-state-metrics,thread-id=*,task-id=*,rocksdb-state-id=*

Real metrics under that MBean include block-cache-data-hit-ratio, block-cache-index-hit-ratio, block-cache-filter-hit-ratio, bytes-written-rate, bytes-read-rate, write-stall-duration-avg, write-stall-duration-total, number-open-files, and number-file-errors-total. (There is no compaction-pending, num-running-flushes, or num-running-compactions metric—those names are not exposed by Kafka Streams.) Sustained, rising write-stall-duration-total is the clearest signal that writes are outpacing compaction.

Restore progress is best observed from the per-partition lag exposed by allLocalStorePartitionLags() (the LagInfo.offsetLag() used in the readiness probe above) and from the embedded restore consumer’s standard consumer metrics. Rebalance churn shows up in stream-thread-metrics (for example task-created-rate, task-closed-rate).

Export with the Prometheus JMX exporter:

lowercaseOutputName: true
rules:
  - pattern: 'kafka.streams<type=stream-state-metrics, thread-id=(.+), task-id=(.+), rocksdb-state-id=(.+)><>(.+):'
    name: kafka_streams_rocksdb_$4
    labels:
      thread_id: "$1"
      task_id: "$2"
      store: "$3"
  - pattern: 'kafka.streams<type=stream-thread-metrics, thread-id=(.+)><>(task-created-rate|task-closed-rate):'
    name: kafka_streams_thread_$2
    labels:
      thread_id: "$1"

Alerting on State Store Health

Useful alerts:

  • Restoration not progressing: per-partition offsetLag from allLocalStorePartitionLags() (or restore-consumer lag) stays high for several minutes—an instance may be stuck restoring and unable to serve.
  • RocksDB write stalls: write-stall-duration-total climbing alongside high write latency means compaction is behind; scale the instance vertically or reduce write throughput.
  • Query latency: instrument the query endpoint with a latency histogram and alert when p99 exceeds your SLO; route around the slow host using standby metadata where possible.

Conclusion

Interactive Queries turn a Kafka Streams application from a pure processing engine into a queryable, stateful read layer, removing the external read database and serving lookups at local-RocksDB latency. The cost is operational discipline: route queries correctly (activeHost().host()/port(), not a phantom activePort()), bound RocksDB memory with a shared LRUCache, gate traffic on real restore progress via allLocalStorePartitionLags(), secure the endpoints, and alert on the metrics that actually exist. Treated with the same reliability, security, and observability standards as any production datastore, the result is a single, cohesive system where stream processing and state serving live together.

Back to Stream Processing