Do I need a cluster to run Kafka Streams?

No. Kafka Streams is a Java library, not a server. Your application is the processing cluster — you scale by running more instances of the same jar, and they coordinate through a consumer group exactly like ordinary consumers do.

Where does the state live?

In a local RocksDB store on each instance, backed by a compacted Kafka topic called a changelog. The local store makes reads fast; the changelog makes them recoverable, since a restarted or relocated instance replays it to rebuild state.

What is the difference between a KStream and a KTable?

A KStream is a sequence of independent events where every record matters. A KTable is a changelog where each record replaces the previous value for that key, so it represents current state. The same topic can be read either way depending on what the data means.

Why does my windowed aggregation emit multiple results per window?

By default Kafka Streams emits an updated result on every input record rather than waiting for the window to close. Use suppress(Suppressed.untilWindowCloses(...)) to emit exactly one final result per window, at the cost of waiting for the grace period to expire.

Stream Processing with Kafka Streams

Kafka Streams processes a topic and writes the result to another topic, with state that survives restarts. It is a library rather than a cluster, which makes it the lightest way to do stateful processing on Kafka.

Setup

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-streams</artifactId>
  <version>3.9.0</version>
</dependency>

Word count, the stateful hello world

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

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

public class WordCount {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-v1");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();

        builder.<String, String>stream("sentences")
               .flatMapValues(line -> Arrays.asList(line.toLowerCase().split("\\W+")))
               .filter((key, word) -> !word.isEmpty())
               .groupBy((key, word) -> word)          // re-key by word, triggers a repartition
               .count(Materialized.as("counts-store"))
               .toStream()
               .peek((word, count) -> System.out.printf("%-10s %d%n", word, count))
               .to("word-counts", Produced.with(Serdes.String(), Serdes.Long()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
        streams.start();
    }
}

Feed it three sentences:

printf 'the quick brown fox\nthe lazy dog\nthe quick dog\n' | \
  bin/kafka-console-producer.sh --topic sentences --bootstrap-server localhost:9092

$ java -jar wordcount.jar
the        1
quick      1
brown      1
fox        1
the        2
lazy       1
dog        1
the        3
quick      2
dog        2

Two things worth reading closely.

Every input record produces output immediately. the is emitted at 1, then 2, then 3 — Kafka Streams is continuous, not batch. There is no “final” count because the stream has no end.

The counts survived across sentences. quick reached 2 because the count from the first sentence was still there. That state is real, and it is stored somewhere.

Where the state lives

bin/kafka-topics.sh --list --bootstrap-server localhost:9092 | grep wordcount

wordcount-v1-counts-store-changelog
wordcount-v1-counts-store-repartition

Kafka Streams created two internal topics on your behalf.

The repartition topic exists because groupBy changed the key — records have to be physically moved so that all occurrences of a word land in the same partition, which is what makes the count correct.

The changelog is compacted and holds the latest count per word. Local state lives in RocksDB under /tmp/kafka-streams/wordcount-v1/; the changelog is what lets a new instance rebuild it.

bin/kafka-console-consumer.sh --topic wordcount-v1-counts-store-changelog \
  --from-beginning --property print.key=true \
  --value-deserializer org.apache.kafka.common.serialization.LongDeserializer \
  --max-messages 4 --bootstrap-server localhost:9092

the	3
quick	2
dog	2
brown	1
Processed a total of 4 messages

Kill the app, delete the local state directory, and restart it. It replays the changelog and resumes at the correct counts rather than starting from zero — the state is durable even though it is stored locally.

Windowed aggregation

Counting since the beginning of time is rarely what you want. Count per five-minute window:

builder.<String, String>stream("clicks")
       .groupByKey()
       .windowedBy(TimeWindows.ofSizeAndGrace(
               Duration.ofMinutes(5), Duration.ofMinutes(1)))
       .count()
       .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
       .toStream()
       .foreach((windowedKey, count) ->
           System.out.printf("%s [%s - %s] %d%n",
               windowedKey.key(),
               windowedKey.window().startTime(),
               windowedKey.window().endTime(),
               count));

$ java -jar clickcount.jar
alice [2026-02-14T10:00:00Z - 2026-02-14T10:05:00Z] 12
bob   [2026-02-14T10:00:00Z - 2026-02-14T10:05:00Z] 3
alice [2026-02-14T10:05:00Z - 2026-02-14T10:10:00Z] 7

One row per key per window. Without suppress you would get an updated row on every single click — correct, but usually far too noisy for a downstream consumer.

The grace period is the second argument: one minute of tolerance for records that arrive late because of a network delay or a slow producer. A record older than window_end + grace is dropped rather than reopening a closed window, and Streams counts those drops in the dropped-records metric. Watch that metric — a nonzero value means you are silently discarding real data.

Joining a stream to a table

The common enrichment shape: events on one side, reference data on the other.

// Reference data — one record per key, latest wins.
KTable<String, String> users =
    builder.table("users", Consumed.with(Serdes.String(), Serdes.String()));

// Events — every record matters.
KStream<String, String> orders =
    builder.stream("orders", Consumed.with(Serdes.String(), Serdes.String()));

orders.join(users, (order, user) -> user + " ordered " + order)
      .peek((key, joined) -> System.out.println(joined))
      .to("orders-enriched");

printf 'u1:Alice\nu2:Bob\n' | bin/kafka-console-producer.sh --topic users \
  --property parse.key=true --property key.separator=: --bootstrap-server localhost:9092
printf 'u1:espresso\nu2:cortado\nu3:latte\n' | bin/kafka-console-producer.sh --topic orders \
  --property parse.key=true --property key.separator=: --bootstrap-server localhost:9092

Alice ordered espresso
Bob ordered cortado

u3 produced nothing. An inner join drops records with no match, and there is no user u3 in the table. Use leftJoin to keep them with a null on the table side:

Alice ordered espresso
Bob ordered cortado
null ordered latte

Both sides must be co-partitioned — same partition count, same keying strategy — because the join happens locally on each instance without any cross-instance lookup. A mismatch gives you a TopologyException at startup rather than silently wrong results, which is the right failure.

Scaling out

Run a second instance of the same jar with the same application.id:

$ java -jar wordcount.jar     # instance 2
INFO Cooperative rebalance: assigned tasks [0_1, 0_2]

and instance 1 logs:

INFO Cooperative rebalance: revoked tasks [0_1, 0_2], retained [0_0]

Tasks — one per input partition — redistributed across the instances, and each instance restores only the state for the tasks it now owns. Scaling is just running more copies of the process; the topology, the state, and the coordination are handled for you.

Practice

1. Change application.id and restart. What happens to the counts?

the        1
quick      1

They restart from zero. The application.id is the consumer group ID and the prefix for the internal topics, so changing it creates an entirely separate application with new changelog topics and no committed offsets. This is the supported way to reprocess a stream from scratch — and an easy way to accidentally lose state.

2. Remove suppress from the windowed count. How many outputs per window?

One per input record. For a window with 12 clicks you get 12 emissions with counts 1 through 12. Downstream consumers that treat each row as final will double-count. Either suppress, or make the consumer idempotent on (key, window_start).

3. Join two topics with different partition counts.

org.apache.kafka.streams.errors.TopologyException: Invalid topology:
Following topics do not have the same number of partitions: [orders: 3, users: 1]

Streams refuses to build the topology. Fix it by recreating one topic with a matching partition count, or by using a GlobalKTable, which replicates the full table to every instance and therefore has no co-partitioning requirement — at the cost of holding the whole table in each instance’s memory.

4. Send a record timestamped 10 minutes in the past to a 5-minute window with 1-minute grace.

It is dropped, silently as far as your code is concerned. Check the metric:

stream-task-metrics:dropped-records-total  1.0

Streams uses the record’s own timestamp, not arrival time, so a producer with a skewed clock or a long backlog can push records outside the grace period. Alert on dropped-records-total — it is the only signal that this is happening.

Next: the settings that actually move throughput and latency once correctness is settled.