Interactive queries in Kafka - Mini Project: Build & Apply

Interactive Queries with Kafka Streams

📖 Scenario: You are building a simple Kafka Streams application that counts the number of purchases per product in real time. You want to be able to query the current count of any product interactively.

🎯 Goal: Create a Kafka Streams application that maintains a count of purchases per product and supports interactive queries to get the current count for a given product.

📋 What You'll Learn

Create a Kafka Streams topology with a KTable counting purchases per product

Configure the application with an application ID

Implement interactive queries to fetch the current count for a product

Print the count result for a specific product

💡 Why This Matters

🌍 Real World

Interactive queries let you ask your Kafka Streams app for the latest computed data without waiting for output topics. This is useful for dashboards and real-time monitoring.

💼 Career

Kafka Streams interactive queries are important for building responsive, stateful stream processing applications in data engineering and backend development roles.

1

Create the initial purchase data stream

Create a Kafka Streams KStream named purchases from the topic purchase-events with key type String and value type String.

Kafka

# Create a KStream named purchases from topic 'purchase-events'
# Your code here

Need a hint?

Use builder.stream("purchase-events") to create the stream.

2

Configure the Kafka Streams application ID

Create a Properties object named props and set the StreamsConfig.APPLICATION_ID_CONFIG property to "purchase-count-app".

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream purchases = builder.stream("purchase-events");

# Create Properties object props and set application ID to 'purchase-count-app'
# Your code here

Need a hint?

Use props.put(StreamsConfig.APPLICATION_ID_CONFIG, "purchase-count-app").

3

Create a KTable counting purchases per product

Use the purchases stream to create a KTable<String, Long> named purchaseCounts that counts the number of purchases per product key using groupByKey() and count().

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream purchases = builder.stream("purchase-events");

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "purchase-count-app");

# Create a KTable purchaseCounts that counts purchases per product key
# Your code here

Need a hint?

Use purchases.groupByKey().count() to create the count table.

4

Print the current count for a product using interactive queries

Start the Kafka Streams application with KafkaStreams streams = new KafkaStreams(builder.build(), props); and streams.start(). Then, use streams.store() with QueryableStoreTypes.keyValueStore() to get the ReadOnlyKeyValueStore<String, Long> named store. Finally, print the count for the product key "product-123" using store.get("product-123").

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream purchases = builder.stream("purchase-events");

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "purchase-count-app");

KTable purchaseCounts = purchases.groupByKey().count();

# Start KafkaStreams, get the store, and print count for 'product-123'
# Your code here

Need a hint?

Use streams.store() with QueryableStoreTypes.keyValueStore() and then store.get("product-123").