Home/Learn/Apache Kafka/Interactive Queries

Interactive Queries

Advanced

Kafka Streams

Interactive Queries expose state store contents over a REST API, turning the Kafka Streams instance into a queryable read model without an external DB.

Overview

Interactive Queries (IQ) allow you to query Kafka Streams local state stores directly from outside the stream topology, turning a Kafka Streams application into a queryable, real-time read model. Rather than writing aggregated state to an external database, you query the local RocksDB-backed stores via the KafkaStreams.store() API. In a multi-instance deployment, each instance holds only the partitions assigned to it. To answer queries for any key, instances must discover each other (via application.server config) and proxy requests to the instance that owns the relevant partition. This is the "distributed queryable state" pattern that eliminates the external DB entirely.

Querying local state stores

Use KafkaStreams.store(StoreQueryParameters) to access a read-only view of a named state store. The store must be materialized with a name — anonymous stores are not queryable. Available store types: keyValueStore() (point lookup + range scan), windowStore() (windowed queries), sessionStore() (session window queries).

Java — materialized state store and REST query endpoint

// 1. Materialize the state store with a name during topology build
StreamsBuilder builder = new StreamsBuilder();

KTable<String, Long> wordCounts = builder
    .stream("text-input", Consumed.with(Serdes.String(), Serdes.String()))
    .flatMapValues(v -> Arrays.asList(v.split("\\s+")))
    .groupBy((k, word) -> word)
    .count(Materialized.as("word-count-store"));  // named → queryable

// 2. Query the store from a REST endpoint
@RestController
public class WordCountController {

    private final KafkaStreams streams;

    @GetMapping("/counts/{word}")
    public Long getCount(@PathVariable String word) {
        ReadOnlyKeyValueStore<String, Long> store = streams.store(
            StoreQueryParameters.fromNameAndType(
                "word-count-store",
                QueryableStoreTypes.keyValueStore())
        );
        Long count = store.get(word);
        return count != null ? count : 0L;
    }

    @GetMapping("/counts")
    public Map<String, Long> getAllCounts() {
        ReadOnlyKeyValueStore<String, Long> store = streams.store(
            StoreQueryParameters.fromNameAndType(
                "word-count-store", QueryableStoreTypes.keyValueStore()));
        Map<String, Long> result = new HashMap<>();
        try (KeyValueIterator<String, Long> it = store.all()) {
            it.forEachRemaining(kv -> result.put(kv.key, kv.value));
        }
        return result;
    }
}

Distributed queries across instances

When running multiple Kafka Streams instances, each holds a subset of partitions (and thus keys). To answer a query for any key, the receiving instance must discover which instance owns the partition for that key, then proxy the request. Configure application.server so instances can discover each other via metadata.

Java — distributed query with partition ownership discovery and proxying

# application.properties — advertise this instance's REST endpoint
spring.kafka.streams.properties.application.server=192.168.1.10:8080

// Find which instance owns a given key
@GetMapping("/counts/{word}")
public Long getCountDistributed(@PathVariable String word) {
    // Find the owner of this key's partition
    KeyQueryMetadata metadata = streams.queryMetadataForKey(
        "word-count-store", word, Serdes.String().serializer());

    HostInfo activeHost = metadata.activeHost();
    String thisHost = "192.168.1.10";
    int thisPort = 8080;

    if (activeHost.host().equals(thisHost) && activeHost.port() == thisPort) {
        // This instance owns the key — query locally
        ReadOnlyKeyValueStore<String, Long> store = streams.store(
            StoreQueryParameters.fromNameAndType(
                "word-count-store", QueryableStoreTypes.keyValueStore()));
        return store.get(word);
    } else {
        // Proxy to the owning instance
        String url = String.format("http://%s:%d/counts/%s",
            activeHost.host(), activeHost.port(), word);
        return restClient.get().uri(url).retrieve().body(Long.class);
    }
}

Windowed store queries

Windowed state stores (from tumbling/hopping/session window aggregations) require a WindowStore or SessionStore query. You can retrieve results for a key within a specific time range. This enables real-time dashboards showing metrics over the last N minutes without a separate TSDB.

Java — windowed state store query for time-ranged aggregations

// Windowed aggregation — count per word per 1-minute tumbling window
KTable<Windowed<String>, Long> windowedCounts = builder
    .stream("text-input", Consumed.with(Serdes.String(), Serdes.String()))
    .flatMapValues(v -> Arrays.asList(v.split("\\s+")))
    .groupBy((k, word) -> word)
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
    .count(Materialized.as("windowed-word-count"));

// Query windowed store — get counts for "kafka" in last 5 minutes
@GetMapping("/windowed-counts/{word}")
public List<Map<String, Object>> getWindowedCounts(@PathVariable String word) {
    ReadOnlyWindowStore<String, Long> windowStore = streams.store(
        StoreQueryParameters.fromNameAndType(
            "windowed-word-count", QueryableStoreTypes.windowStore()));

    Instant now = Instant.now();
    Instant from = now.minus(Duration.ofMinutes(5));

    List<Map<String, Object>> results = new ArrayList<>();
    try (WindowStoreIterator<Long> it = windowStore.fetch(word, from, now)) {
        it.forEachRemaining(kv -> results.add(Map.of(
            "windowStart", Instant.ofEpochMilli(kv.key),
            "count", kv.value
        )));
    }
    return results;
}

Key Points to Remember

1State stores must be materialized with a name (Materialized.as("name")) to be queryable — anonymous stores are not accessible
2streams.store(StoreQueryParameters.fromNameAndType(...)) returns a read-only view of the local state store
3In multi-instance deployments, each instance owns only its assigned partitions — use queryMetadataForKey to find the owner
4Configure application.server=host:port so instances can discover each other for distributed query proxying
5WindowStore supports time-ranged fetch queries — retrieve aggregations for a key over a specific time range
6Interactive Queries eliminate the need to sink aggregated state to an external database for simple read models

Interview Questions

What does a Kafka Streams state store need to be queryable via Interactive Queries?

EasyConfluent

In a 3-instance Kafka Streams deployment, how does an instance answer a query for a key it does not own?

HardLinkedIn

What is the purpose of the application.server configuration in Kafka Streams?

MediumUber

How would you query a windowed state store to retrieve per-minute counts for the last hour?

HardNetflix

What are the trade-offs of using Interactive Queries as a read model vs sinking state to a database?

MediumAmazon

Ask Aria about Interactive Queries

Your personal AI tutor — ask anything about this concept

Revision Status

Personal Notes

Discussion

Loading discussion…

State Stores & RocksDB

Kafka Connect