Interactive queries in Kafka - Commands & Configuration

Introduction

Interactive queries let you ask a running Kafka Streams application for the current state of its data. This helps you get real-time answers without waiting for data to be stored elsewhere.

When you want to check the current count of events processed by your streaming app.

When you need to get the latest value of a key from your Kafka Streams state store.

When you want to build a dashboard that shows live data from your streaming application.

When you want to avoid querying an external database and instead ask the streaming app directly.

When you want to combine streaming processing with fast, on-demand data lookups.

Commands

Create a Kafka topic named 'pageviews' with 3 partitions to hold streaming data.

Terminal

kafka-topics --create --topic pageviews --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

Expected Output

Created topic pageviews.

→

--partitions - Number of partitions for parallelism

→

--replication-factor - Number of copies for fault tolerance

Start a producer to send sample pageview events to the 'pageviews' topic.

Terminal

kafka-console-producer --topic pageviews --bootstrap-server localhost:9092

Expected Output

No output (command runs silently)

→

--topic - Topic to send messages to

→

--bootstrap-server - Kafka server address

Query the Kafka Streams app's REST endpoint to get the current state for key '123' from the 'pageviews' store.

Terminal

curl http://localhost:8080/state/pageviews/123

Expected Output

{"key":"123","count":42}

Query the state store for a key '999' that may not exist to see how the app responds.

Terminal

curl http://localhost:8080/state/pageviews/999

Expected Output

{"key":"999","count":0}

Key Concept

Interactive queries let you ask a live Kafka Streams app for its current data state directly, enabling real-time insights without extra storage.

Common Mistakes

Trying to query a state store without enabling interactive queries in the Kafka Streams app.

The app won't expose the state store, so queries will fail or return errors.

Configure the Kafka Streams app with interactive queries enabled and expose a REST endpoint.

Querying a key that does not exist without handling empty or default responses.

You may get null or errors, causing your app or UI to crash.

Handle missing keys gracefully by returning default values like zero or empty objects.

Not matching the topic and store names correctly between the Kafka Streams app and the query URL.

The query will not find the correct store and fail.

Ensure the topic and state store names used in queries exactly match those in the Kafka Streams app.

Summary

Create a Kafka topic to hold streaming data.

Send data to the topic using a producer.

Query the Kafka Streams app's REST endpoint to get live state data for specific keys.

Handle missing keys and ensure the app is configured for interactive queries.

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