06 - Kafka Streams: Stream Processing Embedded in Java, KTable and Windowing

Counting each user's orders every five minutes, inside your service

You have a Java service that already reads orders from a topic. You are asked to show, in real time, how many orders each user has placed in the last five minutes, and to enrich each order with the product name. You can write a loop with a consumer, an in-memory map and a database so as not to lose the counts on restart. Or you can use Kafka Streams: a library that is added to your application and runs the stream-processing logic inside your own processes, without a separate cluster. The documentation says it in one clear sentence: Kafka Streams "is not a resource manager, but a library that runs anywhere its stream processing application runs".

The library gives you four tools. The stream (KStream) and the table (KTable), two ways of reading the same topic. Local state with its recovery log. Time windows. Joins between stream and table. In this article you see them really run: I used kafka-streams-4.4.0.jar and the TopologyTestDriver of kafka-streams-test-utils (Maven Central, SHA-1 verified), which runs the topology in memory, one record at a time, without a broker. That is why the article shows real behaviours of the library, not an invented model; it does not show behaviours that require a cluster (rebalance, state restoration after a failure), which I describe from the 4.4 documentation consulted on 2026-11-12. The stores are in memory (dsl.store.suppliers.class set to the in-memory ones) so as not to depend on RocksDB, the default on-disk engine.

Topology, tasks and threads: where parallelism lives

The code you write describes a topology: a graph of operators (filters, transformations, aggregations) connected by streams of records. The library breaks it into tasks, a fixed number decided by the partitions of the input topics: each task receives a set of partitions and that assignment never changes. From this derives the rule that holds for every Streams application: the maximum parallelism is the number of input partitions. With a topic of five partitions you can start up to five instances; the sixth waits and takes over if one fails. It is the same rule as the consumer groups in the article on the fundamentals, because the same mechanism sits underneath.

Inside an instance, tasks run on threads (num.stream.threads, default 1) with no shared state. The listing prints from the real library the defaults that matter for a new application.

import org.apache.kafka.streams.StreamsConfig;
import java.util.*;

public class StreamsDefaults {
    public static void main(String[] args) {
        Map<String, Object> p = new HashMap<>();
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-analytics");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");   // not contacted
        StreamsConfig c = new StreamsConfig(p);
        for (String k : List.of("num.stream.threads", "num.standby.replicas", "commit.interval.ms",
                "statestore.cache.max.bytes", "state.dir", "deserialization.exception.handler",
                "processing.exception.handler", "production.exception.handler", "errors.dead.letter.queue.topic.name"))
            System.out.println(k + " = " + c.values().get(k));
    }
}

// Output:
// num.stream.threads = 1
// num.standby.replicas = 0
// commit.interval.ms = 30000
// statestore.cache.max.bytes = 10485760
// state.dir = /tmp/kafka-streams
// deserialization.exception.handler = class org.apache.kafka.streams.errors.LogAndFailExceptionHandler
// processing.exception.handler = class org.apache.kafka.streams.errors.LogAndFailProcessingExceptionHandler
// production.exception.handler = class org.apache.kafka.streams.errors.DefaultProductionExceptionHandler
// errors.dead.letter.queue.topic.name = null

Four lines deserve attention. num.standby.replicas = 0 means that after a failure a task's state must be rebuilt by rereading its changelog, and therefore that recovery takes time proportional to the amount of state; standby replicas shorten it at the cost of memory and disk. commit.interval.ms = 30000 applies to the at-least-once guarantee: with exactly_once_v2 it drops to 100 (article on transactions). state.dir = /tmp/kafka-streams is the default path, computed from the system's temporary directory: the documentation warns that it must be unique for every instance sharing the filesystem, and on many systems /tmp is emptied at reboot, so in production set a persistent path. The last three are the error handlers: two fail (LogAndFail...) and one applies the default behaviour. We will come back to errors.dead.letter.queue.topic.name, which is null by default.

KStream and KTable: two readings of the same topic

An order goes from PENDING to CONFIRMED to SHIPPED: three records with the same key. If you care about every change, you read the topic as a KStream: each record is an independent event. If you care about the current state, you read it as a KTable: each record with the same key replaces the previous one. The documentation calls this link "stream-table duality": a stream is the register of changes of a table, and a table is a snapshot of a stream. The compacted topic of the article on the fundamentals exploits the same idea.

import org.apache.kafka.common.serialization.*;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.BuiltInDslStoreSuppliers;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.common.utils.Bytes;
import java.util.*;

public class StreamVsTable {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        // The same kind of data read in two ways.
        KStream<String, String> events = builder.stream("order-states", Consumed.with(Serdes.String(), Serdes.String()));
        events.groupByKey().count(Materialized.as("events-per-order")).toStream()
              .to("counts", Produced.with(Serdes.String(), Serdes.Long()));
        builder.table("order-states-table", Consumed.with(Serdes.String(), Serdes.String()),
                      Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("current-state"));

        Properties p = new Properties();
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "demo");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");         // never contacted
        p.put(StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG, BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class);
        p.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);           // one output for every update

        try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), p)) {
            TestInputTopic<String, String> in = driver.createInputTopic("order-states",
                new StringSerializer(), new StringSerializer());
            TestInputTopic<String, String> inTab = driver.createInputTopic("order-states-table",
                new StringSerializer(), new StringSerializer());
            for (String state : List.of("PENDING", "CONFIRMED", "SHIPPED")) {
                in.pipeInput("order-123", state);
                inTab.pipeInput("order-123", state);
            }
            TestOutputTopic<String, Long> out = driver.createOutputTopic("counts",
                new StringDeserializer(), new LongDeserializer());
            System.out.println("KStream: count updates for order-123 -> " + out.readValuesToList());
            KeyValueStore<String, String> state = driver.getKeyValueStore("current-state");
            System.out.println("KTable: current state of order-123 -> " + state.get("order-123"));
        }
    }
}

// Output:
// KStream: count updates for order-123 -> [1, 2, 3]
// KTable: current state of order-123 -> SHIPPED

The first branch counts events per key: the output is [1, 2, 3], one update for each event. The second branch is a table and keeps only the last value: SHIPPED. The line that disables the cache (statestore.cache.max.bytes = 0) is there only for the demonstration: with the default 10 MiB the library merges several updates of the same key before emitting them, and you would see fewer lines in the output. An aggregation (count, aggregate) always produces a KTable, even when starting from a stream: it is what allows the library to correct the value if an out-of-order record arrives.

The state of an aggregation lives in a local state store. So as not to lose it with the machine, each store has a changelog: a compacted Kafka topic, partitioned like the store, in which the library records every update. If the task restarts on another machine, it rereads the changelog and rebuilds the store ("Architecture" documentation). It is not all free: each aggregation creates one more topic, and the restart time depends on its size.

Windows: counting by time intervals

A window groups records by the event timestamp, not by the moment they arrive. The three main forms: the tumbling has fixed, non-overlapping windows (0-5, 5-10 minutes); the hopping has fixed-size windows that advance by a smaller step and therefore overlap; the session groups the events of the same user separated by less than an inactivity gap. There is also the sliding one, which follows each event. Every window has a grace period: the documentation explains that a record is dropped if its timestamp assigns it to a window but the stream's current time is beyond the end of the window plus the grace.

import org.apache.kafka.common.serialization.*;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.BuiltInDslStoreSuppliers;
import java.time.*;
import java.util.*;

public class Windows {
    static final long MIN = 60_000L;

    static KeyValue<String, Long> label(Windowed<String> w, Long n) {
        return KeyValue.pair(w.key() + " [" + w.window().start() / MIN + "," + w.window().end() / MIN + ")", n);
    }

    public static void main(String[] args) {
        StreamsBuilder b = new StreamsBuilder();
        KGroupedStream<String, String> perUser = b.<String, String>stream("orders").groupByKey();

        perUser.windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1)))
            .count().toStream().map(Windows::label).to("tumbling", Produced.with(Serdes.String(), Serdes.Long()));
        perUser.windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(10), Duration.ofMinutes(1))
                .advanceBy(Duration.ofMinutes(5)))
            .count().toStream().map(Windows::label).to("hopping", Produced.with(Serdes.String(), Serdes.Long()));
        perUser.windowedBy(SessionWindows.ofInactivityGapAndGrace(Duration.ofMinutes(30), Duration.ofMinutes(1)))
            .count().toStream().map(Windows::label).to("sessions", Produced.with(Serdes.String(), Serdes.Long()));

        Properties p = new Properties();
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "demo-windows");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
        p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        p.put(StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG, BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class);
        p.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);

        try (TopologyTestDriver driver = new TopologyTestDriver(b.build(), p)) {
            TestInputTopic<String, String> in = driver.createInputTopic("orders",
                new StringSerializer(), new StringSerializer());
            // minutes of the event time: 1, 3, 7, 40 and then a late arrival with time 2, arriving after minute 40
            for (long minute : new long[] {1, 3, 7, 40, 2})
                in.pipeInput("u1", "order", Instant.ofEpochMilli(minute * MIN));

            for (String topic : List.of("tumbling", "hopping", "sessions")) {
                TestOutputTopic<String, Long> out = driver.createOutputTopic(topic,
                    new StringDeserializer(), new LongDeserializer());
                System.out.println(topic + ": " + new TreeMap<>(out.readKeyValuesToMap()));
            }
        }
    }
}

// Output:
// tumbling: {u1 [0,5)=2, u1 [40,45)=1, u1 [5,10)=1}
// hopping: {u1 [0,10)=3, u1 [35,45)=1, u1 [40,50)=1, u1 [5,15)=1}
// sessions: {u1 [1,1)=null, u1 [1,3)=null, u1 [1,7)=3, u1 [40,40)=1}

The events of user u1 have time (in minutes) 1, 3, 7 and 40, and the last one arrives with time 2: the late arrival. In the 5-minute tumbling the window [0,5) counts 2 (minutes 1 and 3) and not 3, because the late arrival came when the stream time was 40, well beyond the end of the window plus the one-minute grace, and it is dropped. In the 10-minute hopping with step 5 minute 7 appears in two windows, [0,10) (which counts 3) and [5,15); minute 40 in two others. In the sessions, with a maximum inactivity of 30 minutes, minutes 1, 3 and 7 form a single session [1,7) of 3 events, and 40 opens another. The rows with value null are the intermediate sessions ([1,1) and [1,3)) that merged into the larger one: the library emits a tombstone for each.

Two warnings. If you want a result only when the window closes, and not every update, the documentation points to suppress(Suppressed.untilWindowCloses(...)). And if you set a grace that is too small you lose data silently, too large you delay emission: the choice depends on the maximum delay your producers can accumulate, and must be measured on your stream.

Enriching a stream with a table

The most frequent case is enriching events with reference data: the order with the product name. In Streams it is a join between a KStream (the orders) and a KTable (the catalog), which requires something the documentation calls co-partitioning: the two topics must have the same number of partitions and the same key partitioning strategy, because the join happens on records with the same key in the same task. The exception is the join with a GlobalKTable, which has a full copy on every instance and does not require co-partitioning: it costs, though, disk space and network traffic, because every instance reads the whole topic.

import org.apache.kafka.common.serialization.*;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.BuiltInDslStoreSuppliers;
import java.util.*;

public class Enrichment {
    public static void main(String[] args) {
        StreamsBuilder b = new StreamsBuilder();
        // Orders: key = product id, value = order id. Catalog: key = product id, value = name.
        KStream<String, String> orders = b.stream("orders", Consumed.with(Serdes.String(), Serdes.String()));
        KTable<String, String> catalog = b.table("catalog", Consumed.with(Serdes.String(), Serdes.String()));

        orders.leftJoin(catalog, (order, product) -> order + " -> " + (product == null ? "unknown product" : product))
              .to("enriched", Produced.with(Serdes.String(), Serdes.String()));

        Properties p = new Properties();
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "demo-join");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
        p.put(StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG, BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class);

        try (TopologyTestDriver driver = new TopologyTestDriver(b.build(), p)) {
            TestInputTopic<String, String> inOrders = driver.createInputTopic("orders", new StringSerializer(), new StringSerializer());
            TestInputTopic<String, String> inCatalog = driver.createInputTopic("catalog", new StringSerializer(), new StringSerializer());

            inOrders.pipeInput("p-9", "order-1");               // the product is not in the catalog yet
            inCatalog.pipeInput("p-9", "Mechanical keyboard");  // now it is
            inOrders.pipeInput("p-9", "order-2");

            TestOutputTopic<String, String> out = driver.createOutputTopic("enriched", new StringDeserializer(), new StringDeserializer());
            out.readValuesToList().forEach(System.out::println);
        }
    }
}

// Output:
// order-1 -> unknown product
// order-2 -> Mechanical keyboard

The first order arrives when the catalog does not yet have the product: the leftJoin lets it through and the product value is null, so "unknown product" comes out. After the catalog record arrives, the second order is enriched correctly. The lesson is that the stream-table join is not retroactive: order 1 is not re-emitted when the table is updated, because the join is triggered only by the arrival of a record from the stream side. An inner join would have dropped order 1; the left join keeps it with the placeholder. Choose based on what is less serious: losing an order or emitting it with missing data.

Errors and state queries

The error handler for records that cannot be deserialized is, by default, LogAndFailExceptionHandler: the application stops. It is the prudent choice, because a corrupt record silently discarded is lost data; LogAndContinueExceptionHandler skips it and continues. In recent versions the configuration has a novelty: errors.dead.letter.queue.topic.name, which is null by default. If you set it, the default handler builds a record for the dead letter queue and sends it to that topic. It does not apply to global stores, and it is ignored if you have set a handler of your own. The detail of the discard patterns is in the article on dead letter queues.

Local state can also be queried from outside the topology (the interactive queries, with KafkaStreams.store(...)); I show no listing because it requires an instance started against a broker. The idea to take away is that each instance knows only the partitions it manages: to answer about the whole state you need to know which instance owns the key.

What you can do now

You know that Kafka Streams is a library and that its maximum parallelism is the number of input partitions. You can choose between reading as a stream and as a table and know why an aggregation always produces a table. You can predict which events enter a tumbling, hopping or session window and why a late arrival can be dropped. You know that the stream-table join is not retroactive and that it requires co-partitioning. And you remember to set state.dir to a persistent path. When pipelines become analytical across several streams and with independent scaling, the comparison is with Apache Flink; to bring data into and out of Kafka without writing code, with Kafka Connect. If you arrive from another point of the series, the links are: the Schema Registry.