Kafka in Action 05 - Deploying and Querying Your Application

We have reached the final chapter of our “Kafka in Action” series. Over the last four articles, we have assembled a complete developer’s toolkit for Kafka Streams. We’ve built pipelines with stateless transformations, performed complex stateful aggregations within time windows, and enriched event streams by joining them with reference data. We now have the knowledge to build almost any real-time data processing topology.

But writing the code is only half the battle. A real application must be deployed, scaled, and monitored. And what if we want to interact with our application’s results directly, without having to read from a Kafka topic?

This concluding article will bridge the gap from development to production. We will discuss deployment and scalability, and then introduce a paradigm-shifting feature that transforms your Streams application from a data pipeline into a queryable microservice: Interactive Queries.

Deploying and Scaling a Streams Application

One of the most appealing aspects of Kafka Streams is its operational simplicity.

It’s Just a Java Application.

Unlike cluster-based processing frameworks like Apache Spark or Flink, a Kafka Streams application is not a “job” that you submit to a special cluster. It is a standard Java application packaged in a JAR file. This means you can run and deploy it using the tools you already know:

• On bare-metal servers or VMs by running java -jar your-app.jar.
• As a containerized service using Docker.
• On an orchestration platform like Kubernetes.

Elastic, Horizontal Scalability

How do you make your application process data faster? You simply start more instances of it.

When you start a new instance of your application with the same application.id, it joins the existing consumer group. As we learned in our “Hello Kafka” series, this triggers a rebalance. Kafka Streams automatically pauses processing, re-distributes the topic partitions (and their associated state) among all available instances, and resumes.

This makes scaling incredibly simple:

• To scale up: Deploy more instances.
• To scale down: Shut down instances.

The framework handles the partition and state migration for you, making your application truly elastic and fault-tolerant. If an instance fails, its workload is automatically shifted to the remaining healthy instances.

The Problem: Getting Data Out

So far, the only way we’ve retrieved results from our applications is by sinking them to an output topic and reading that topic with a console consumer. This is a perfect pattern for event-driven systems where downstream applications need to react to the results.

But what if a user or a front-end service needs to ask a direct question? “What is the current sales total for ‘Laptops’ right now?” or “What is the current count for the word ‘kafka’?” Writing the result to a topic and waiting for a consumer to read it is slow and inefficient for this kind of point-in-time, request/response query.

The Solution: Interactive Queries

Interactive Queries allow you to directly query the state stores that your Kafka Streams application maintains locally. When you perform a stateful operation like count(), aggregate(), or a join(), Streams materializes the state in a local, fault-tolerant key-value store (typically RocksDB). Interactive Queries let you expose this store to the outside world, for example, over a REST API.

This effectively turns your stream processing application into a fully-fledged, queryable microservice that serves real-time, materialized views of your data streams.

Building a Queryable Word Count Service

Let’s revisit our very first WordCountApp and add a REST endpoint to it. We want to be able to make a GET request to /count/{word} and get back the current count for that word.

1. Add Dependencies

We need a lightweight web server library. We’ll add Javalin to our pom.xml.

<dependency>
    <groupId>io.javalin</groupId>
    <artifactId>javalin</artifactId>
    <version>5.6.1</version> <!-- Use a recent version -->
</dependency>

2. Modify the WordCountApp

We need to make two key changes:

1. Give our state store a name so we can reference it later. We do this using Materialized.as("store-name").
2. Embed and start the Javalin web server after starting the KafkaStreams client.

import io.javalin.Javalin;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;

import java.util.Arrays;
import java.util.Properties;

public class QueryableWordCountApp {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "queryable-word-count-app");
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        // Required for exposing the store via an RPC endpoint
        properties.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "localhost:7000");

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> textLines = builder.stream("word-count-input");

        // ★★★ 1. Give the state store a queryable name: "word-counts"
        KTable<String, Long> wordCounts = textLines
                .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
                .groupBy((key, word) -> word)
                .count(Materialized.as("word-counts"));

        wordCounts.toStream().to("word-count-output", Produced.with(Serdes.String(), Serdes.Long()));

        KafkaStreams streams = new KafkaStreams(builder.build(), properties);
        streams.start();

        // ★★★ 2. Start a web server to expose the state store
        Javalin app = Javalin.create().start(7000);
        
        app.get("/count/{word}", ctx -> {
            String word = ctx.pathParam("word");
            
            // Get the handle to the local state store
            ReadOnlyKeyValueStore<String, Long> keyValueStore = 
                streams.store(StoreQueryParameters.fromNameAndType("word-counts", QueryableStoreTypes.keyValueStore()));

            Long count = keyValueStore.get(word);

            if (count != null) {
                ctx.json("Count for '" + word + "': " + count);
            } else {
                ctx.status(404).json("Word not found.");
            }
        });

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            streams.close();
            app.stop();
        }));
    }
}

Running the Service

1. Start the App: Run the QueryableWordCountApp.
2. Produce Data: Start a console producer and send some text to the word-count-input topic.

   docker-compose exec kafka kafka-console-producer --broker-list localhost:9092 --topic word-count-input
   >hello kafka
   >kafka is great
   

3. Query the Service: Open your web browser or use curl to query the endpoint.

   curl http://localhost:7000/count/kafka
   # Output: "Count for 'kafka': 2"
   
   curl http://localhost:7000/count/great
   # Output: "Count for 'great': 1"
   
   curl http://localhost:7000/count/unknown
   # Output: "Word not found."
   

   You are now directly querying the state of your real-time stream processing application!

A Note on Distributed State: In this simple example, we are querying the state store on a single instance. In a scaled-out deployment, the data for a key will only exist on the one instance that is managing that key’s partition. A production-ready application would need logic to discover which instance hosts the key and redirect the HTTP request to the correct instance. Kafka Streams provides the necessary metadata APIs to build this routing logic.

Conclusion of the Series

Across this series, we have seen that Kafka Streams is far more than a simple library; it is a complete framework for building modern, real-time applications and microservices. We’ve journeyed from the basics of stateless transformations to the complexities of stateful, windowed aggregations, contextual joins, and finally, direct state querying.

You now possess the fundamental knowledge to design, build, and deploy sophisticated stream processing applications that are scalable, fault-tolerant, and operationally simple. The world of real-time data is at your fingertips. Happy streaming