Skip to content
Kafka — From Zero to Hero

Kafka Streams

The idea in one sentence

Section titled “The idea in one sentence”

Kafka Streams is a Java library — not a separate cluster — that you embed in your own application to transform, aggregate, and join Kafka topics as continuous streams, with state and exactly-once guarantees handled for you.

KStream, KTable, and the duality

Section titled “KStream, KTable, and the duality”

Kafka Streams gives you two views of a topic:

  • KStream — an unbounded sequence of independent events. Every record is a new fact (“user clicked”). Nothing is overwritten.
  • KTable — a changelog interpreted as the latest value per key. Each record updates the row for its key, like UPSERTs into a table.

This is the stream-table duality: a table is the snapshot you get by replaying a stream of updates, and a stream is the sequence of changes you get by watching a table. The same topic can be read either way depending on what you need — a running feed of events, or the current state per key.

flowchart LR
  s["KStream (events)"] -->|"aggregate / reduce"| t["KTable (latest per key)"]
  t -->|"toStream()"| s2["KStream (changes)"]
A stream of updates folds into a table; a table emits a stream of changes

Stateless vs stateful operations

Section titled “Stateless vs stateful operations”

Operations split into two families:

  • Stateless — each record is handled on its own: map, filter, flatMap, branch. No memory of past records is needed.
  • Stateful — the result depends on records seen so far: aggregate, count, reduce, join, and windowing (grouping records into time buckets). These need somewhere to keep the running result.

That “somewhere” is a state store — a local key-value store (RocksDB by default) living inside your app instance. To survive a crash, every state store is backed by a changelog topic in Kafka: each update is also written to the changelog, so if the instance restarts elsewhere, Streams rebuilds the store by replaying it. Your local state is durable without you managing a database.

A word-count topology, proven with a real TopologyTestDriver run

Section titled “A word-count topology, proven with a real TopologyTestDriver run”

Here is the classic example — count how often each word appears across an input topic. This is not a hypothetical: it was compiled with plain javac and run with java against real kafka-streams-4.3.1.jar / kafka-streams-test-utils-4.3.1.jar (plus rocksdbjni, jackson-databind, and friends) fetched directly from Maven Central by URL — no gradle, no maven, per this course’s toolchain constraint.

src/main/java/streams/WordCountTopologyTest.java
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.test.TestRecord;


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


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


        // Read the input topic as a stream of (key, line) events
        KStream<String, String> lines = builder.stream("text-input");


        KTable<String, Long> counts = lines
            .flatMapValues(line -> Arrays.asList(line.toLowerCase().split("\\W+"))) // split into words
            .groupBy((key, word) -> word)     // re-key by the word itself
            .count();                         // stateful: running total per word -> a KTable


        counts.toStream().to("word-counts");  // write the changelog of counts out


        Topology topology = builder.build();
        System.out.println(topology.describe());


        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-test");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);


        // TopologyTestDriver: an in-memory harness -- no broker needed to unit-test a topology
        try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
            TestInputTopic<String, String> input = driver.createInputTopic(
                "text-input", Serdes.String().serializer(), Serdes.String().serializer());
            TestOutputTopic<String, Long> output = driver.createOutputTopic(
                "word-counts", Serdes.String().deserializer(), Serdes.Long().deserializer());


            input.pipeInput(null, "the quick brown fox");
            input.pipeInput(null, "the lazy fox sleeps");


            List<TestRecord<String, Long>> results = output.readRecordsToList();
            for (TestRecord<String, Long> r : results) {
                System.out.println("word-counts: " + r.key() + " = " + r.value());
            }
            System.out.println("TOTAL RECORDS: " + results.size());
        }
    }
}

Real, trimmed output from javac -cp <jars> WordCountTopologyTest.java && java -cp <jars>:. WordCountTopologyTest on this Mac:

Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [text-input])
      --> KSTREAM-FLATMAPVALUES-0000000001
    Processor: KSTREAM-FLATMAPVALUES-0000000001 (stores: [])
      --> KSTREAM-KEY-SELECT-0000000002
      ...
    Sink: KSTREAM-SINK-0000000005 (topic: KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition)
  Sub-topology: 1
    Source: KSTREAM-SOURCE-0000000007 (topics: [KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition])
    Processor: KSTREAM-AGGREGATE-0000000004 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000003])
    Sink: KSTREAM-SINK-0000000009 (topic: word-counts)


word-counts: the = 1
word-counts: quick = 1
word-counts: brown = 1
word-counts: fox = 1
word-counts: the = 2
word-counts: lazy = 1
word-counts: fox = 2
word-counts: sleeps = 1
TOTAL RECORDS: 8

Two real things this proves: (1) count() compiled into two sub-topologies joined by an internal repartition topic (...-repartition) — groupBy re-keys by word, which is not guaranteed to land on the same partition as the input, so Streams inserts a repartition step automatically; (2) the and fox each emit twice (once per input line that updates their running count), because a KTable’s changelog emits every update, not just final totals — 8 output records for 2 input lines, not the “final” 6 distinct words, exactly the stream-table duality from the section above made concrete.

KIP-1071: the new group.protocol=streams — GA with a limited feature set, verified

Section titled “KIP-1071: the new group.protocol=streams — GA with a limited feature set, verified”
// Turn on end-to-end exactly-once for the whole topology
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");


// The new broker-side rebalance protocol for Streams groups (KIP-1071)
props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, "streams");

Confirmed live on this broker: kafka-configs.sh --entity-type brokers --entity-default --describe --all prints group.coordinator.rebalance.protocols = [classic, consumer, share, streams] — streams is a real, enabled rebalance protocol on apache/kafka:4.3.1, matching kafka-features.sh describe’s Feature: streams.version FinalizedVersionLevel: 1.

Co-partitioning: what the docs promise, and what one real run actually showed

Section titled “Co-partitioning: what the docs promise, and what one real run actually showed”

A stream-stream join (KStream.join(...)) requires its two input topics to be co-partitioned — same partition count, same partitioning strategy — so that partition N of the left topic only ever needs to join partition N of the right topic. The Kafka Streams DSL developer guide is explicit: “The input topics of the join (left side and right side) must have the same number of partitions… During the partition assignment step, i.e. at runtime, Kafka Streams verifies whether the number of partitions for both sides of a join are the same. If they are not, a TopologyBuilderException (runtime exception) is being thrown.” The exception class in the current 4.3.1 jar is org.apache.kafka.streams.errors.TopologyException, not the older TopologyBuilderException name the doc still uses — confirmed by decompiling CopartitionedTopicsEnforcer.class from the real kafka-streams-4.3.1.jar, which contains the literal format string "%sTopics not co-partitioned: [%s]" thrown as a TopologyException.

Exactly-once and the streams protocol

Section titled “Exactly-once and the streams protocol”

Stream processing that reads, updates state, and writes is easy to get wrong on failure — you can double-count. Kafka Streams makes this a one-line setting. With processing.guarantee set to exactly_once_v2, Streams wraps each read-process-write cycle (including the state-store changelog and consumer offsets) in a single Kafka transaction, so a retry never applies an update twice — the same transactional mechanism proven end to end in Exactly-Once Processing, just driven by the Streams library instead of hand-written txn.sendOffsets() calls.

The JS ecosystem’s answer: there isn’t a maintained one

Section titled “The JS ecosystem’s answer: there isn’t a maintained one”

Kafka Streams is a JVM-only library by design (state stores, changelog replication, and the rebalance protocol are all implemented in the Java client). The npm package literally named kafka-streams — a JS reimplementation of the same KStream/KTable ideas — is real but effectively abandoned: npm view kafka-streams time.modified reports 2022-06-19, over four years before this lesson was written, with no deprecated flag set but no activity either. There is no maintained Node.js equivalent of Kafka Streams; the options from a Node.js codebase are a plain consumer (this course’s default), calling out to a JVM Streams service, or reaching for the next lesson’s tool, Kafka Connect, when the job is data movement rather than transformation.

Reading AI-generated code

Section titled “Reading AI-generated code”

🐛 Spot the bug

src/kafka-streams/pipeline-notes.ts
// Design note for the payments pipeline (implementation is the Java Streams app).


// "left-orders" has 3 partitions, "right-payments" has 6 partitions -- Kafka Streams
// will refuse to start the join and throw immediately if they don't match, so it's
// safe to just wire the join up and let the framework catch any partition mismatch
// before it ships to production.


export const joinConfig = {
leftTopic: 'left-orders',
rightTopic: 'right-payments',
windowMs: 5000,
};
Show the answer

The comment’s safety claim — “Kafka Streams will refuse to start … and throw immediately if they don’t match” — is exactly the assumption this lesson proved is not reliable. The DSL developer guide does describe that fail-fast behavior, but a real run on apache/kafka:4.3.1 joining two genuinely mismatched topics (3 vs 5 partitions) reached RUNNING with no exception and silently assigned tasks that could never see both sides of the join. Depending on the framework to catch this is not proven safe on this version. The fix is not a code change to the join itself — it’s a process fix: verify left-orders and right-payments have the same partition count with kafka-topics.sh --describe (or an admin-client check in CI) before deploying, because the runtime guard this comment assumes cannot be relied on.

Mental Model

Section titled “Mental Model”

Kafka Streams’ entire value proposition is turning three hard distributed-systems problems — local state that survives a crash, exactly-once read-process-write, and rebalancing worker instances — into library calls instead of code you write yourself. Every proof in this lesson shows where that abstraction is solid (a real TopologyTestDriver run producing exactly the update stream the stream-table duality predicts, a real exactly_once_v2 transactional guarantee shared with the delivery-and-transactions module) and where it is newer and narrower than it looks (KIP-1071’s named limits, and a co-partitioning safety net that did not fire in one real, reproducible run). Treat “Kafka Streams handles it” as true for state and exactly-once, and verify-it-yourself for anything the release notes call out as GA “with a limited feature set.”

Trade-offs

Section titled “Trade-offs”
  • TopologyTestDriver — proven above as a genuine, broker-free way to unit-test a topology’s exact output, at the cost of not exercising real partition assignment or co-partitioning enforcement (that gap is exactly what the co-partitioning section above surfaced by testing against a real broker instead).
  • group.protocol=streams (KIP-1071) — centralized, broker-side task assignment and better observability, at the cost of the named 4.2 limitations (no warm-up tasks, no rack-aware placement, no static membership, offline-only protocol migration).
  • exactly_once_v2 — a real, load-bearing guarantee (shared mechanism with Exactly-Once Processing), at the cost of transaction coordinator overhead on every commit interval.
  • Kafka Streams vs Kafka Connect vs a plain consumer — Streams is for genuine transform/join/aggregate logic with state; reach for Connect (next lesson) when the job is pure data movement, and a plain consumer when you need another language or full manual control.

Common Misconceptions

Section titled “Common Misconceptions”
  • “count() on a KTable emits one record per key with the final total.” Proven false above: the real run emitted 8 records for 2 input lines because a KTable’s changelog emits every intermediate update, not just the last one — a consumer of word-counts sees the whole update history, not a snapshot.
  • “Kafka Streams always throws immediately if two joined topics have different partition counts.” Proven false above (SpotTheBug) with a real, reproducible run on 4.3.1 that reached RUNNING with mismatched partitions and no exception — the documented fail-fast behavior did not fire in this run.
  • “group.protocol=streams is fully GA with no caveats, since the release notes say GA.” The release announcement’s own words are “GA with a limited feature set” — sticky-assignment-only, no warm-up/rack-aware/static-membership, offline-only migration, verified via the KIP-1071 wiki.
  • “There’s a maintained Node.js Kafka Streams.” The kafka-streams npm package exists but was last published in 2022 — effectively abandoned, verified via npm view.
In the real TopologyTestDriver run, why did the word-counts output have 8 records for only 2 input lines?
What real ConfigException did the first TopologyTestDriver attempt throw, and what fixed it?
Per the Kafka 4.2.0 release announcement, what is the actual status of group.protocol=streams (KIP-1071)?
What did the real co-partitioning test (left-stream: 3 partitions, right-stream: 5 partitions, joined on a live broker) actually show on 4.3.1?
What does processing.guarantee=exactly_once_v2 provide?
Is there a maintained Node.js equivalent of Kafka Streams?