60-Day Kafka 4 Learning Plan · Week 5— Day 29 of 60

60-Day Kafka 4 Learning Plan · Week 5 — Kafka Streams Sources: Kafka: The Definitive Guide Ch.11 · kafka.apache.org/documentation/streams

Goal

Understand what Kafka Streams is, how KStream and KTable differ at a fundamental level, what’s actually happening under the hood with RocksDB and changelog/repartition topics, and how to wire up, test, and monitor a basic Streams topology in Spring Boot 3 with Kafka 4.

1 What is Kafka Streams?

Kafka Streams is a Java library for real-time stateful stream processing embedded directly in your application. Unlike Flink or Spark, it requires no separate cluster — it runs as threads inside your Spring Boot process.

Key properties:

✅ No separate cluster needed — runs inside your app
✅ Exactly-once semantics (Kafka 4 KRaft makes this more reliable)
✅ Fault-tolerant stateful aggregations via RocksDB + changelog topics
✅ Scales horizontally — add more instances, Kafka re-partitions the load

2 KStream vs KTable — the core mental model

KStream — unbounded event stream

Every record is an independent event. Records are appended and never replaced. Think of it like a newspaper — each edition is a new, separate fact.

Topic: labs.events
(key="u1", value=ORDER_PLACED)   ← event 1
(key="u1", value=ORDER_RETURNED) ← event 2

KStream sees → both records, emits both downstream
Result → 2 rows, not 1 updated row

Use for: clicks, purchases, sensor readings, log lines — any fact that happened exactly once.

KTable — materialized changelog view

Each key holds only its latest value. A new record for the same key is an upsert (replaces the previous value). Think of it like a database row — the latest state wins.

Topic: labs.user-tier
(key="u1", value=SILVER)  ← record 1
(key="u1", value=GOLD)    ← record 2

KTable sees → only GOLD for "u1"
Result → 1 row, latest value

Use for: user profiles, inventory levels, running totals — any current-state lookup.

The tombstone pattern

A null value means different things:

KStream → a deletion event (publish to signal something was removed)
KTable → a tombstone (removes the key from the materialized view entirely)

3 Streams topology — source → processor → sink

A topology is a directed acyclic graph of stream processors. Kafka Streams builds it from your DSL code at startup.

Source (labs.events)
  → Filter (status == CONFIRMED)
  → Map (key = userId)
  → Sink (labs.confirmed)
         ↘ groupByKey → count → KTable (labs.order-counts)

Every processor in the topology runs in-process. Kafka Streams handles partitioning, threading, and state distribution automatically.

4 DSL code — KStream filter + KTable aggregation

// OrderStreamsApp.java — Spring Boot + Kafka Streams 4
@Configuration
public class OrderStreamsApp {

    @Bean
    KStream<String, OrderEvent> confirmedOrders(StreamsBuilder builder) {

        // § Source — read from Avro topic
        KStream<String, OrderEvent> orders = builder.stream("labs.events");

        // § Filter + Map — only CONFIRMED, key = userId
        KStream<String, OrderEvent> confirmed = orders
            .filter((key, value) -> value.getStatus() == Status.CONFIRMED)
            .selectKey((key, value) -> value.getUserId().toString());

        // § KTable — order count per user (materialized view in RocksDB)
        KTable<String, Long> countPerUser = confirmed
            .groupByKey()
            .count(Materialized.as("order-counts"));

        // § Sink — write KTable changelog to output topic
        countPerUser.toStream().to("labs.order-counts");

        return confirmed;
    }
}

Key DSL operators

5 Spring Boot application.yml

spring:
  kafka:
    streams:
      application-id: labs-streams-app   # unique per logical app — used as consumer group ID
      bootstrap-servers: localhost:9092
      default-deserialization-exception-handler: LogAndContinueExceptionHandler
      properties:
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
        schema.registry.url: http://localhost:8081
        processing.guarantee: exactly_once_v2   # Kafka 4 exactly-once

Required Maven dependency

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-streams</artifactId>
</dependency>

6 KStream vs KTable quick-reference

7 Kafka Streams 4 changes from Kafka 3

KRaft-only: no ZooKeeper dependency simplifies Streams metadata management
Streams DSL v4: cleaner API surface, deprecated methods removed
Improved changelog compaction: KTable state restores faster after restarts
Task assignment: improved rack-awareness for partition assignment
processing.guarantee: exactly_once_v2: the recommended EOS mode (replaces exactly_once)

8 RocksDB state stores & changelog topics — what’s really happening

Materialized.as("order-counts") in §4 isn’t just an in-memory map — it creates a local RocksDB instance on disk (one per partition of the aggregation, distributed across your app’s instances) and a compacted changelog topic (labs-streams-app-order-counts-changelog, using Day 9’s compaction semantics) that backs it up continuously.

Every update to the "order-counts" state store:
  1. Written to local RocksDB (fast, disk-backed, survives process restart)
  2. Written to the changelog topic (compacted — latest value per key retained)

On instance restart or rebalance to a new node:
  → RocksDB is rebuilt by replaying the changelog topic from the beginning
  → Restore time scales with changelog size, same pattern as KRaft snapshot
    replay (Day 8 §5) — this is why "improved changelog compaction" in
    Kafka Streams 4 (§7) directly reduces restore/rebalance time

Practical implications:

The changelog topic needs disk planning like any compacted topic (Day 9 §9) — it’s not free, and it grows with unique key cardinality, not event volume.
Local disk on the instance running Streams needs to be large enough for the full RocksDB state — this is a real capacity planning input, not an afterthought, for any aggregation over a large key space.
Standby replicas (num.standby.replicas) pre-warm a second copy of the state store on another instance, so a rebalance can fail over without a full changelog replay — trading extra disk/network for much faster recovery.

spring:
  kafka:
    streams:
      properties:
        num.standby.replicas: 1   # one warm standby per state store partition

9 Repartition topics — the hidden topic `selectKey` creates

selectKey() in §4 changes the record’s key — but downstream operations (like groupByKey) need records for the same new key to land on the same partition, which the original topic’s partitioning can’t guarantee once the key has changed. Kafka Streams handles this transparently by creating an internal repartition topic.

labs.events (partitioned by original key)
   → selectKey(newKey)
   → INTERNAL repartition topic (labs-streams-app-...-repartition, partitioned by NEW key)
   → groupByKey (now safe — same key guaranteed same partition)

Why this matters operationally: repartition topics are real Kafka topics — they consume disk, need monitoring, and add a network hop (write + re-read) that has real latency/throughput cost. A topology with several selectKey/map-then-groupByKey chains can create multiple hidden repartition topics without it being obvious from reading the DSL code alone. Use the topology description (topology.describe()) to see exactly which internal topics your app creates before assuming a topology is “free” just because it’s expressed in a few lines of DSL.

System.out.println(topology.describe()); // shows every internal topic explicitly

10 Threading model & scaling

Each Streams app instance runs one or more StreamThreads, each owning a set of tasks (a task = one partition’s worth of the topology, including its local state store shard).

spring:
  kafka:
    streams:
      properties:
        num.stream.threads: 3   # threads within THIS instance

Max useful num.stream.threads follows the same partition-count ceiling as consumer concurrency (Day 17 §1) — more threads than input partitions sit idle.
Scaling out means running more application instances, not just raising num.stream.threads on one — Kafka Streams rebalances tasks (and their state) across all running instances of the same application-id, exactly like a consumer group rebalance (Day 6 §4), because under the hood a Streams app is a consumer group.
application-id doubles as the consumer group ID — this is why it must be globally unique per logical Streams application; two unrelated apps sharing an application-id would compete for the same partitions and corrupt each other’s state stores.

11 Testing with TopologyTestDriver

TopologyTestDriver runs the topology in-memory without a real Kafka cluster or @EmbeddedKafka — much faster for unit-testing topology logic itself.

class OrderStreamsTopologyTest {

    private TopologyTestDriver testDriver;
    private TestInputTopic<String, OrderEvent> inputTopic;
    private TestOutputTopic<String, Long> outputTopic;

    @BeforeEach
    void setup() {
        StreamsBuilder builder = new StreamsBuilder();
        // ... build the same topology as OrderStreamsApp ...
        Topology topology = builder.build();

        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
        testDriver = new TopologyTestDriver(topology, props);

        inputTopic = testDriver.createInputTopic("labs.events",
            new StringSerializer(), avroSerializer);
        outputTopic = testDriver.createOutputTopic("labs.order-counts",
            new StringDeserializer(), new LongDeserializer());
    }

    @Test
    void countsOnlyConfirmedOrdersPerUser() {
        inputTopic.pipeInput("k1", confirmedOrder("u1"));
        inputTopic.pipeInput("k2", pendingOrder("u1"));  // filtered out
        inputTopic.pipeInput("k3", confirmedOrder("u1"));

        assertThat(outputTopic.readValue()).isEqualTo(1L);
        assertThat(outputTopic.readValue()).isEqualTo(2L); // second CONFIRMED increments
    }

    @AfterEach
    void tearDown() {
        testDriver.close(); // releases the in-memory RocksDB instance
    }
}

TopologyTestDriver vs @EmbeddedKafka: use TopologyTestDriver for fast, focused topology logic tests (this is the Streams equivalent of MockProducer from Day 16 §7 — testing your own logic, not real broker interaction). Reserve @EmbeddedKafka for genuine end-to-end integration tests that need real consumer group rebalancing, real serialization over the wire, or multiple cooperating applications.

12 Monitoring a Streams app

# The application-id IS the consumer group — inspect it exactly like any other
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group labs-streams-app

Because a Streams app is a consumer group under the hood, every monitoring habit from Day 6 §7 (lag tracking) and Day 17 §9 applies directly — kafka-consumer-groups.sh --describe works exactly the same way here as for any @KafkaListener consumer.

13 Common pitfalls

Sizing local disk for the app instance without accounting for RocksDB state — a large-key-cardinality aggregation needs real disk capacity planning, not an afterthought (§8)
Not realizing selectKey/map-then-groupByKey creates a hidden repartition topic — invisible in casual DSL reading, real in disk/network cost; use topology.describe() to see it (§9)
Raising num.stream.threads past the input partition count — same wasted-thread pattern as consumer concurrency (Day 17 §1); extra threads sit idle
Reusing an application-id across genuinely different logical apps — corrupts state store assignment since it’s literally the consumer group ID (§10)
Testing only with TopologyTestDriver and never validating with @EmbeddedKafka — topology logic can be correct in isolation while still having real integration issues (serde config, actual partitioning) that only a real-broker test surfaces (§11)

Key Takeaways

Kafka Streams runs inside your app — no separate Flink/Spark cluster needed
KStream = every event matters (append); KTable = only the latest state per key (upsert)
A topology is a DAG: source → processors (filter, map, join) → sink
KTable state is backed by RocksDB and a compacted changelog topic — restore time on rebalance scales with changelog size, same pattern as KRaft snapshot replay
selectKey/key-changing operations before groupByKey create hidden internal repartition topics — inspect with topology.describe()
A Streams app IS a consumer group — application-id is the group ID, and all the usual consumer group monitoring/scaling rules apply directly
num.standby.replicas trades disk/network for faster rebalance recovery by pre-warming state store copies
TopologyTestDriver for fast topology unit tests; @EmbeddedKafka for real end-to-end integration
spring.kafka.streams.application-id must be unique per logical streams app
null value on a KTable key = tombstone; use it to delete entries from the materialized view

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next post.

➡️ Day 30: Topology DSL — filter, map, flatMap, branch

Resources

📘 Kafka: The Definitive Guide — Chapter 11 (Kafka Streams)
🌐 kafka.apache.org/documentation/streams
🌐 docs.spring.io/spring-kafka/reference/streams.html
🌐 Kafka Streams — TopologyTestDriver

👉 Link to Medium blog