Querying State Stores with REST APIs in Kafka Streams

In Stream Processing architectures, the internal state of a running application is often the most useful operational signal you have. When you build Stateful Stream Processing with Kafka Streams applications, the local state stores hold real-time aggregations and key-value lookups that are valuable for debugging, monitoring, and lightweight internal dashboards. Reaching that state from outside the JVM requires a deliberate design, because state is partitioned across instances. This guide shows how to wrap Kafka Streams interactive queries in a REST layer correctly, including the cluster-aware routing that a naive implementation gets wrong.

Understanding Interactive Queries and RPC Patterns

The KafkaStreams instance exposes state through the store() method, which returns a read-only handle to a local state store. The operational detail that drives the entire design: state is sharded across instances by partition assignment, so a given key may live on a different host than the one that received the HTTP request.

The standard pattern has two steps. First, call queryMetadataForKey() to discover which instance owns the key. If the key is local, serve it directly; if it is remote, forward the request (an HTTP redirect, or an internal RPC fan-out). Two facts about queryMetadataForKey() matter for correctness and are easy to get wrong:

  • It returns null when no metadata can be found for the store/key, not a sentinel object.
  • It returns the static sentinel KeyQueryMetadata.NOT_AVAILABLE when metadata is temporarily unavailable, typically during a rebalance.

You must guard against both. Comparing against NOT_AVAILABLE alone (a mistake in many examples) throws a NullPointerException the moment the method returns null. Always put the constant on the left of equals() so the null case is handled safely.

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyQueryMetadata;
import org.apache.kafka.streams.StoreQueryParameters;
import org.apache.kafka.streams.state.HostInfo;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.common.serialization.Serdes;

public class StoreQueryService {
    private final KafkaStreams streams;
    private final HostInfo currentHost;

    public StoreQueryService(KafkaStreams streams, HostInfo host) {
        this.streams = streams;
        this.currentHost = host;
    }

    public String getKeyValue(String storeName, String key) {
        KeyQueryMetadata metadata = streams.queryMetadataForKey(
            storeName, key, Serdes.String().serializer()
        );

        // queryMetadataForKey returns null if no metadata exists, and the
        // NOT_AVAILABLE sentinel during a rebalance. Handle both.
        if (metadata == null || KeyQueryMetadata.NOT_AVAILABLE.equals(metadata)) {
            throw new InvalidStateStoreException(
                "Metadata not available; the store may be initializing or rebalancing."
            );
        }

        if (metadata.activeHost().equals(currentHost)) {
            ReadOnlyKeyValueStore<String, String> store = streams.store(
                StoreQueryParameters.fromNameAndType(
                    storeName, QueryableStoreTypes.keyValueStore()
                )
            );
            return store.get(key); // may be null if the key is absent
        }
        // Remote host — forward via RPC. metadata.activeHost() gives host/port.
        throw new RemoteHostException(metadata.activeHost());
    }
}

RemoteHostException here is your own exception that carries the target HostInfo so the caller can redirect or proxy; the REST controller below shows that handling inline instead.

Cross-instance discovery only works if every instance advertises its RPC endpoint. Set application.server (the StreamsConfig.APPLICATION_SERVER_CONFIG property) to the host:port of this instance’s REST API before starting the stream:

props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "10.0.1.5:8080");

Without it, activeHost() returns the unknown host HostInfo("unavailable", -1) for every key, and routing silently breaks.

Designing the REST API Layer

The REST layer should be thin: it owns no state and delegates everything to the KafkaStreams instance. Embed a lightweight HTTP server—Jetty, Netty, or Spring Boot’s embedded Tomcat—inside the Kafka Streams application so the server shares the JVM lifecycle with the stream threads.

A production endpoint must handle three realities: the store can be unavailable during a rebalance, the key can reside on a remote instance, and the store type (key-value, windowed, or session) dictates the query shape. The following Spring controller implements a robust key-value lookup with redirect-based routing:

@RestController
@RequestMapping("/api/v1/stores")
public class StateStoreController {

    private final KafkaStreams streams;
    private final HostInfo currentHost;

    public StateStoreController(KafkaStreams streams,
                                @Value("${server.host}") String host,
                                @Value("${server.port}") int port) {
        this.streams = streams;
        this.currentHost = new HostInfo(host, port);
    }

    @GetMapping("/{storeName}/keys/{key}")
    public ResponseEntity<?> getByKey(@PathVariable String storeName,
                                      @PathVariable String key) {
        try {
            KeyQueryMetadata metadata = streams.queryMetadataForKey(
                storeName, key, Serdes.String().serializer()
            );

            if (metadata == null || KeyQueryMetadata.NOT_AVAILABLE.equals(metadata)) {
                return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE)
                    .body(Map.of("error", "Metadata unavailable; likely rebalancing"));
            }

            if (!metadata.activeHost().equals(currentHost)) {
                String redirectUrl = String.format(
                    "http://%s:%d/api/v1/stores/%s/keys/%s",
                    metadata.activeHost().host(),
                    metadata.activeHost().port(),
                    storeName, key
                );
                return ResponseEntity.status(HttpStatus.TEMPORARY_REDIRECT)
                    .header(HttpHeaders.LOCATION, redirectUrl)
                    .build();
            }

            ReadOnlyKeyValueStore<String, String> store = streams.store(
                StoreQueryParameters.fromNameAndType(
                    storeName, QueryableStoreTypes.keyValueStore()
                )
            );
            String value = store.get(key);
            if (value == null) {
                return ResponseEntity.notFound().build();
            }
            return ResponseEntity.ok(Map.of("key", key, "value", value));
        } catch (InvalidStateStoreException e) {
            return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE)
                .body(Map.of("error", "Store is rebalancing, retry shortly"));
        }
    }
}

Two choices are worth calling out. First, the InvalidStateStoreException catch maps the brief window of unavailability during partition reassignment to a 503, which lets client retry logic handle the transient state. Second, this uses 307 Temporary Redirect rather than 302 Found. A 302 permits clients to switch the method to GET; 307 guarantees the original method and body are preserved, which is the correct semantics if you later add write-style query endpoints. For read-only GET traffic either works, but 307 is the safer default.

Handling Remote Queries and Service Discovery

The redirect pattern handles point lookups. Many operational tasks need a fan-out instead—for example, a range scan or a “where does this key live across all replicas” query that must touch every instance and merge the results.

Use streamsMetadataForStore() to enumerate every instance hosting a store. (The older allMetadataForStore() was deprecated in Kafka 3.0 in favor of streamsMetadataForStore(); likewise allMetadata() was replaced by metadataForAllStreamsClients(), and the metadata type now lives in org.apache.kafka.streams.StreamsMetadata, not the deprecated …streams.state.StreamsMetadata.)

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import org.springframework.http.ResponseEntity;

public class RemoteQueryClient {
    private static final Logger logger = LoggerFactory.getLogger(RemoteQueryClient.class);
    private final KafkaStreams streams;
    private final RestTemplate restTemplate;

    public RemoteQueryClient(KafkaStreams streams, RestTemplate restTemplate) {
        this.streams = streams;
        this.restTemplate = restTemplate;
    }

    public Map<String, String> queryAllInstances(String storeName, String key) {
        Collection<StreamsMetadata> allMetadata = streams.streamsMetadataForStore(storeName);
        Map<String, String> results = new HashMap<>();

        for (StreamsMetadata metadata : allMetadata) {
            String url = String.format("http://%s:%d/api/v1/stores/%s/keys/%s",
                metadata.hostInfo().host(), metadata.hostInfo().port(), storeName, key);
            try {
                ResponseEntity<Map> response = restTemplate.getForEntity(url, Map.class);
                if (response.getBody() != null && response.getBody().get("value") != null) {
                    results.put(metadata.hostInfo().toString(),
                                response.getBody().get("value").toString());
                }
            } catch (RestClientException e) {
                logger.warn("Failed to query {}", metadata.hostInfo(), e);
            }
        }
        return results;
    }
}

To avoid an endpoint redirecting to itself, the peer endpoints you fan out to should serve the local store directly rather than re-running discovery. A common pattern is to expose an internal ?local=true variant (or a separate path) that skips the activeHost() check and reads the local store, so the coordinating instance does the routing and peers just answer for their own partitions. The Apache Kafka interactive-queries developer guide documents the discovery APIs in detail.

Configuring State Stores for External Access

A store is only queryable by an externally referenced name if you give it one. In the DSL, supply a name through Materialized.as(); in the Processor API, name the StoreBuilder. Without an explicit name the store gets an internal, generated name that is awkward to discover and not stable across topology changes.

StreamsBuilder builder = new StreamsBuilder();

KStream<String, Order> orders = builder.stream("orders");

orders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
    .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("hourly-order-counts")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long()));

The store name "hourly-order-counts" is the identifier used in the REST path. Windowed stores need a query shape different from key-value stores: the read takes a key plus a time range, and fetch() returns a WindowStoreIterator<V> of <timestamp, value> pairs whose iterator keys are Long window-start timestamps in epoch milliseconds.

@GetMapping("/{storeName}/windows/{key}")
public ResponseEntity<?> getWindowed(
        @PathVariable String storeName,
        @PathVariable String key,
        @RequestParam long from,
        @RequestParam long to) {

    ReadOnlyWindowStore<String, Long> store = streams.store(
        StoreQueryParameters.fromNameAndType(
            storeName, QueryableStoreTypes.windowStore()
        )
    );

    try (WindowStoreIterator<Long> iterator = store.fetch(
            key, Instant.ofEpochMilli(from), Instant.ofEpochMilli(to))) {

        List<Map<String, Object>> windows = new ArrayList<>();
        while (iterator.hasNext()) {
            KeyValue<Long, Long> entry = iterator.next();
            windows.add(Map.of(
                "windowStart", entry.key,   // epoch millis
                "count", entry.value
            ));
        }
        return ResponseEntity.ok(windows);
    }
}

The try-with-resources block closes the WindowStoreIterator; this is mandatory because the iterator holds a RocksDB resource and leaking it leaks native memory. Note this endpoint queries the local instance only—for a complete picture across partitions it must be combined with the discovery/fan-out logic above.

Operational Hardening and Monitoring

Exposing state over HTTP adds failure modes that your SRE team must monitor on both the stream application and the embedded server. A readiness check that probes store availability is the cheapest high-value addition:

@GetMapping("/health")
public ResponseEntity<?> health() {
    Map<String, String> storeStatus = new HashMap<>();
    boolean healthy = true;

    Set<String> localStores = streams.metadataForAllStreamsClients().stream()
            .filter(m -> m.hostInfo().equals(currentHost))
            .flatMap(m -> m.stateStoreNames().stream())
            .collect(Collectors.toSet());

    for (String storeName : localStores) {
        try {
            streams.store(StoreQueryParameters.fromNameAndType(
                storeName, QueryableStoreTypes.keyValueStore()
            ));
            storeStatus.put(storeName, "READY");
        } catch (InvalidStateStoreException e) {
            storeStatus.put(storeName, "REBALANCING");
            healthy = false;
        }
    }

    HttpStatus status = healthy ? HttpStatus.OK : HttpStatus.SERVICE_UNAVAILABLE;
    return ResponseEntity.status(status).body(storeStatus);
}

This uses metadataForAllStreamsClients() (the non-deprecated replacement for allMetadata()) and filters to this instance’s own stores via hostInfo(), so the probe reflects local readiness rather than the whole cluster. Wire it to a Kubernetes readiness probe so traffic is withheld from an instance that cannot serve queries, and use a liveness probe with a longer tolerance so the pod is not killed during an extended rebalance. Also check streams.state() for KafkaStreams.State.RUNNING or REBALANCING; treat ERROR and PENDING_SHUTDOWN as unhealthy.

Security matters as much as availability. The embedded server should never face an untrusted network directly. Bind it to a private interface and terminate TLS and authentication at a reverse proxy or service mesh sidecar. The OWASP REST Security Cheat Sheet covers the relevant controls.

Performance Considerations and Caching

Every store.get() or range read hits RocksDB, the default state store backend. RocksDB is fast, but a hot query path competes with the stream threads for I/O and page cache. For frequently read keys, an application-level cache in front of the store reduces RocksDB reads. Caffeine is a good fit:

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import java.util.concurrent.TimeUnit;

public class CachingStoreQueryService {
    private final Cache<String, String> cache;
    private final StoreQueryService delegate;

    public CachingStoreQueryService(StoreQueryService delegate) {
        this.delegate = delegate;
        this.cache = Caffeine.newBuilder()
            .maximumSize(10_000)
            .expireAfterWrite(30, TimeUnit.SECONDS)
            .build();
    }

    public String getKeyValue(String storeName, String key) {
        String cacheKey = storeName + ":" + key;
        String cached = cache.getIfPresent(cacheKey);
        if (cached != null) {
            return cached;
        }
        String value = delegate.getKeyValue(storeName, key);
        if (value != null) {
            cache.put(cacheKey, value);
        }
        return value;
    }
}

Caching trades freshness for load. The state store itself is continuously updated by stream processing, so any TTL you add is a staleness budget on top of the store’s own consistency. Size the TTL to your tolerance: for a dashboard that refreshes every 60 seconds, a 30-second TTL roughly halves the read load while keeping displayed values within one refresh of ground truth. Do not cache reads that operators use to confirm a just-written value.

Troubleshooting Common Failure Scenarios

InvalidStateStoreException. This is the most common error and is expected during normal operation: the store is not yet initialized, is migrating, or is being closed. Retry with backoff and only alert if it persists well beyond your expected rebalance time. Calling store() returns a handle that can also start throwing this exception mid-use if the partition migrates, so wrap individual reads, not just store acquisition.

Serde mismatches. The serializer you pass to queryMetadataForKey() and the serdes you read with must match the serdes the store was built with. A mismatch produces a ClassCastException or silently wrong/null results. Pin this down with integration tests that write through the topology and read back via the REST path.

RPC hangs on a partitioned peer. A fan-out can stall on an unresponsive instance. Set explicit timeouts; SimpleClientHttpRequestFactory.setConnectTimeout(int) and setReadTimeout(int) both take milliseconds:

SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
factory.setConnectTimeout(2000);  // milliseconds
factory.setReadTimeout(5000);     // milliseconds
RestTemplate restTemplate = new RestTemplate(factory);

These bound the blast radius of one slow peer. Pair them with a circuit breaker (for example Resilience4j) so a chronically failing instance is skipped rather than retried on every request.

Embedded server resource limits. Each concurrent request buffers a request and serializes a response. Cap the server thread pool; operational queries are infrequent relative to stream processing, so 10–20 threads is usually ample, and a bounded pool prevents a query spike from starving the stream threads.

Treated as a first-class operational interface—with correct null/NOT_AVAILABLE handling, cluster-aware routing, the application.server config wired up, readiness probes, bounded RPC timeouts, and TLS at the edge—a state-store REST API turns a Kafka Streams application from a black box into a transparent, debuggable part of your data platform.

Back to Stateful Stream Processing With Kafka Streams