Kafka Streams

offset 1

Mental model

Streams and tables

The same topic can be read in two ways.

They convert into each other. If you replay a stream of updates and keep the latest for each key, you get a table. If you write down every change to a table, you get a stream. The demo at the top shows both at once. A compacted topic (module 01) is a table stored in Kafka.

The program is a graph

You describe steps: read a topic, split, filter, group, count, write a topic. Kafka Streams calls this the topology. The word-count program is short enough to read whole:

StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> lines = builder.stream("streams-plaintext-input");

KTable<String, Long> counts = lines
    .flatMapValues(line -> Arrays.asList(line.toLowerCase().split(" ")))  // one record for each word
    .groupBy((key, word) -> word)                                         // the word becomes the key
    .count();                                                             // a table: word → count

counts.toStream().to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));

Two things happen that the code does not show.

• A repartition topic. groupBy changes the key from "nothing" to "the word". Counting needs all records for one word in one partition. So the library writes the records to an internal topic keyed by word and reads them back. The program is now two parts with a topic between them.
• A changelog topic. count needs a store. Every time a count changes, the library also writes the new value to an internal compacted topic.

Time and windows

"Count for each word" grows for ever. Often you want "count in each minute". A window cuts the stream into pieces by time. The time is the record's own timestamp, the event time, not the clock of the machine that processes it. So a record that arrives late still goes into the window where it belongs.

Joins

A join combines records from two sources that have the same key. The most common one enriches a stream with a table: each order gets the current data of its customer. The table side is a lookup. The stream side drives the output. So the order of arrival matters:

offset 2

How it works internally

Tasks: the unit of work

The library splits the topology into sub-topologies at each repartition topic. For each sub-topology it creates one task for each input partition. A task has an id of two numbers: sub-topology and partition. In the lab, three input partitions and two sub-topologies gave six tasks, and six state directories named 0_0 to 1_2.

A task owns its partition and its piece of the state. Tasks are assigned to stream threads across all running instances by the consumer group. So the partition count of the input topic is the upper limit of parallel work, as for any consumer group.

State stores and changelogs

1. A step that needs memory, such as count, gets a state store. By default this is RocksDB, a key-value store that keeps its data in files on local disk.
2. Each write to the store is also sent to the store's changelog topic, a compacted topic with the same number of partitions as the task's input.
3. On a commit, the library flushes the store, waits for the changelog writes, and commits the input offsets. With exactly-once these are one transaction.
4. When a task moves to an instance that does not have its store, the instance reads the changelog partition from the start and rebuilds the store. This is restoration. The task does not process until it is finished.

In the lab I deleted the local state directory and restarted the program. The next sentence gave "the 6", continuing from 5: the counts came back from the changelog.

Restoration of a large store can take minutes. A standby replica is a second instance that keeps reading the changelog to hold a warm copy, so that a failover is fast. Keeping local disks across restarts has the same effect for planned restarts.

Why output can show fewer updates than you expect

A cache sits in front of each store. Several updates to the same key within one commit interval are merged, and only the last is sent on. The example program sets the cache to zero so that you see every change. With the default 10 MiB cache and 30-second commit interval you would see fewer, later records. The final values are the same.

Late records and the grace period

Records can arrive out of order. A window stays open for late records for its grace period after its end. A record that arrives later than that is dropped and counted in a metric. A longer grace period accepts more late data and makes final results later.

Co-partitioning

For a join, the task for partition 3 of one topic must also have partition 3 of the other, and the same key must be in the same partition on both sides. So both topics need the same number of partitions and the same partitioning by key. The library checks the counts at start and stops if they differ. A GlobalKTable avoids the rule: every instance loads the whole table. Use it for small reference data.

Exactly-once

With processing.guarantee=exactly_once_v2, each commit is a transaction that covers the output records, the changelog records and the input offsets. The commit interval then defaults to 100 milliseconds. unverified: the 100 ms value is from memory

offset 3

Every relevant config

Defaults are from the Kafka 4.3 Kafka Streams configuration reference. A Streams application also accepts every producer and consumer setting from modules 02 and 03.