Interactive queries in Kafka - Step-by-Step Execution

Process Flow - Interactive queries

Start Kafka Streams App

↓

Process Streams & Update Local State

↓

Expose Local State Store

↓

Receive Query Request

↓

Check Local Store for Data

↓

Return Data

↓

Respond to Client with Data

↓

End

The app processes streams and updates local state. When a query comes, it checks local data. If found, returns it; otherwise, asks other instances.

Execution Sample

Kafka

StreamsBuilder builder = new StreamsBuilder();
KTable<String, Long> counts = builder.table("input-topic", Materialized.<String, Long>as("counts-store"));
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();
InteractiveQueryService iqService = streams.interactiveQueryService();
ReadOnlyKeyValueStore<String, Long> store = iqService.getQueryableStore("counts-store", QueryableStoreTypes.keyValueStore());
Long value = store.get("key1");

This code builds a stream table, gets the interactive query service, accesses the local state store, and queries a value by key.

Process Table

Step	Action	Evaluation	Result
1	Start Kafka Streams app	App initializes and starts processing	Streams app running
2	Process input topic and update local state store	Records processed and counts updated	Local store 'counts-store' updated
3	Receive query for key 'key1'	Check local store for 'key1'	Key found locally? Yes
4	Return value for 'key1' from local store	Value retrieved	Value = 42
5	Respond to client with value	Send response	Client receives value 42
6	Query for key 'keyX' not in local store	Check local store for 'keyX'	Key found locally? No
7	Forward query to remote instances	Remote query sent	Remote instance returns value 100
8	Respond to client with remote value	Send response	Client receives value 100
9	End	All queries handled	App continues running

💡 Queries stop when data is found locally or remotely; app keeps running.

Status Tracker

Variable	Start	After Step 2	After Step 4	After Step 7	Final
streams	null	initialized	running	running	running
counts-store	empty	updated with keys	contains key1=42	contains key1=42, keyX unknown	contains key1=42, keyX unknown
query key	none	none	key1	keyX	none
query result	none	none	42	100 (from remote)	none

Key Moments - 3 Insights

Why does the app check the local store first before asking other instances?

Because checking local store is faster and reduces network calls. See execution_table rows 3 and 6 where local check decides next step.

What happens if the key is not found locally or remotely?

The query returns null or empty result. The app tries remote queries after local miss (row 7), but if no instance has data, client gets no data.

How does the app keep the local state updated?

By continuously processing the input topic and updating the local store (row 2). This keeps data fresh for queries.

Visual Quiz - 3 Questions

Look at the execution_table, what is the value returned for key 'key1' at step 4?

A 100

B 42

C null

D key1

💡 Hint

Check row 4 in execution_table where local store returns value 42.

At which step does the app forward the query to remote instances?

A Step 3

B Step 6

C Step 7

D Step 5

💡 Hint

Look at execution_table row 7 where remote query is sent after local miss.

If the local store had the key 'keyX', how would the execution_table change?

A Step 6 would return value locally, skipping steps 7 and 8

B Step 7 would still forward query to remote

C Step 4 would return null

D Step 3 would forward query to remote

💡 Hint

Refer to variable_tracker and execution_table rows 3-8 about local vs remote query handling.

Concept Snapshot

Interactive Queries in Kafka Streams:
- Streams app maintains local state stores.
- Queries first check local store for data.
- If data missing locally, query forwards to remote instances.
- Enables fast, scalable state access in distributed apps.
- Use InteractiveQueryService to access stores programmatically.

Full Transcript

This visual execution trace shows how Kafka Streams interactive queries work. The app starts and processes streams, updating a local state store. When a query arrives, it first checks the local store. If the key is found, it returns the value immediately. If not found, it forwards the query to other instances to find the data remotely. The app responds to the client with the found value. Variables like the streams app state, local store contents, query keys, and results change step-by-step. Key moments clarify why local checks happen first, what if data is missing, and how state updates. The quiz tests understanding of values returned, query forwarding steps, and how local data presence affects flow. The snapshot summarizes the concept for quick recall.

Practice

1.

What is the main purpose of interactive queries in Kafka Streams?

easy

A. To create new Kafka topics dynamically

B. To write data back to Kafka topics

C. To read live state data directly from the Kafka Streams application

D. To monitor Kafka broker health status

Solution

1. Step 1: Understand interactive queries concept

   Interactive queries allow reading live state from Kafka Streams state stores.

2. Step 2: Identify the main use case

   The main use is to get real-time data from the app without extra databases.

3. Final Answer:

   To read live state data directly from the Kafka Streams application -> Option C

4. Quick Check:

   Interactive queries = read live state [OK]

Hint: Interactive queries read live state, not write or monitor [OK]

Common Mistakes:

• Confusing interactive queries with writing data
• Thinking it creates topics
• Assuming it monitors brokers

2.

Which of the following is the correct way to get a state store named myStore of type ReadOnlyKeyValueStore in Kafka Streams?

easy

A. store = streams.store("myStore", ReadOnlyKeyValueStore.class);

B. store = streams.getStore("myStore", ReadOnlyKeyValueStore);

C. store = streams.getStore("myStore", ReadOnlyKeyValueStore.class);

D. store = streams.store("myStore", ReadOnlyKeyValueStore);

Solution

1. Step 1: Recall Kafka Streams store retrieval syntax

   The correct method is streams.store(storeName, storeType.class) in Java.

2. Step 2: Match the correct syntax

   store = streams.store("myStore", ReadOnlyKeyValueStore.class); uses streams.store with class type, which is correct method name and syntax.

3. Final Answer:

   store = streams.store("myStore", ReadOnlyKeyValueStore.class); -> Option A

4. Quick Check:

   Use streams.store with class type [OK]

Hint: Use streams.store("name", StoreClass.class) syntax [OK]

Common Mistakes:

• Using streams.getStore instead of streams.store
• Omitting .class for type
• Passing type without .class

3.

Given the following code snippet in a Kafka Streams app, what will be the output if the key "user123" exists with value "active" in the state store?

var store = streams.store("userStatusStore", ReadOnlyKeyValueStore.class);
var status = store.get("user123");
System.out.println(status);

medium

A. null

B. "active"

C. Throws NullPointerException

D. Compilation error

Solution

1. Step 1: Understand store.get(key) behavior

   The get method returns the value for the key if it exists, else null.

2. Step 2: Check given key and value

   Since "user123" exists with value "active", store.get("user123") returns "active".

3. Final Answer:

   "active" -> Option B

4. Quick Check:

   store.get(existingKey) = value [OK]

Hint: store.get(key) returns value or null if missing [OK]

Common Mistakes:

• Assuming null is returned even if key exists
• Expecting exceptions on missing keys
• Confusing syntax causing compile errors

4.

Identify the error in this Kafka Streams interactive query code snippet:

var store = streams.store("ordersStore", ReadOnlyKeyValueStore.class);
var order = store.get(12345);
System.out.println(order);

medium

A. Passing integer key instead of string key

B. Missing try-catch block for exceptions

C. Using streams.store instead of streams.getStore

D. Incorrect store type ReadOnlyKeyValueStore

Solution

1. Step 1: Check method to get state store

   The method streams.store is correct.

2. Step 2: Validate key type and store type

   Key type mismatch: passing integer 12345 instead of string key (common for ordersStore).

3. Final Answer:

   Passing integer key instead of string key -> Option A

4. Quick Check:

   Key type must match store definition [OK]

Hint: Use correct key type in store.get(key) [OK]

Common Mistakes:

• Passing int literal without quotes for string keys
• Assuming key type must be string always
• Ignoring exception handling (not mandatory here)

5.

You want to build a real-time dashboard that shows the current count of active users using Kafka Streams interactive queries. Which approach best fits this requirement?

hard

A. Write user statuses to a separate database and query it periodically

B. Store counts in a local variable updated by Kafka Streams processors

C. Use Kafka consumer to read all user status events and aggregate externally

D. Use a ReadOnlyKeyValueStore to store user statuses and query counts live

Solution

1. Step 1: Identify real-time state storage method

   Kafka Streams state stores hold live data accessible via interactive queries.

2. Step 2: Choose best approach for live dashboard

   Using ReadOnlyKeyValueStore allows querying current counts directly without extra DB.

3. Final Answer:

   Use a ReadOnlyKeyValueStore to store user statuses and query counts live -> Option D

4. Quick Check:

   Interactive queries + state store = real-time dashboard [OK]

Hint: Use state store with interactive queries for live data [OK]

Common Mistakes:

• Relying on external DB instead of state store
• Polling Kafka consumer instead of querying store
• Using local variables which are not shared state