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

---

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

Goal

Master the core stateless DSL operators in Kafka Streams 4: how each transforms the stream, when each triggers an internal repartition, how to control repartitioning explicitly, how exceptions inside these operators actually behave, and how to test and monitor a stateless topology.

1. filter — keep matching records

Cardinality: 1 → 0 or 1. Each input record either passes or is dropped.

// filter: keep records matching predicate
KStream<String, OrderEvent> confirmed = orders
    .filter((key, value) -> value.getStatus() == Status.CONFIRMED);

// filterNot: inverse predicate shorthand (avoids .filter(... !predicate))
KStream<String, OrderEvent> notConfirmed = orders
    .filterNot((key, value) -> value.getStatus() == Status.CONFIRMED);

No repartition — key is unchanged; Kafka Streams keeps the same partition assignment.

2. map & mapValues — transform records

Cardinality: 1 → 1. Each input becomes exactly one output.

// map: change BOTH key and value — causes internal repartition
KStream<String, String> rekeyed = orders
    .map((key, value) -> KeyValue.pair(
        value.getUserId().toString(),
        value.getOrderId().toString()
    ));

// mapValues: transform value only — key unchanged, NO repartition ✅
KStream<String, Double> amounts = orders
    .mapValues(value -> value.getAmount());

// mapValues with key access (BiFunction variant)
KStream<String, String> summary = orders
    .mapValues((key, value) ->
        String.format("user=%s order=%s", key, value.getOrderId()));

⚠️ map() changes the key → triggers an automatic internal repartition topic. This adds latency, network traffic, and storage overhead. Always prefer mapValues() when the key doesn’t need to change.

3. flatMap — one record → many records

Cardinality: 1 → N. Each input can produce 0, 1, or many outputs.

KStream<String, LineItem> lineItems = orders
    .flatMap((key, order) -> order.getItems().stream()
        .map(item -> KeyValue.pair(item.getItemId(), item))
        .toList());

// flatMapValues — same but key unchanged, NO repartition ✅
KStream<String, String> tags = orders
    .flatMapValues(order -> order.getTags());  // one order → N tag strings

Use case: Exploding batch events into individual records, expanding arrays, generating multiple notifications from one trigger event.

Note: flatMap() changes the key (triggers repartition). flatMapValues() keeps the key — prefer it when possible.

4. split/branch — route to multiple streams

Cardinality: 1 → 1 (routed). Each record goes to exactly one branch based on predicate evaluation order.

// Kafka 4 split().branch() API — replaces deprecated KStream.branch() array
Map<String, KStream<String, OrderEvent>> branches = orders
    .split(Named.as("tier-"))
    .branch(
        (key, value) -> value.getAmount() > 500,
        Branched.as("high")
    )
    .branch(
        (key, value) -> value.getAmount() > 100,
        Branched.as("medium")
    )
    .defaultBranch(Branched.as("low"));   // catches everything else

// Access each branch by name
KStream<String, OrderEvent> highValue  = branches.get("tier-high");
KStream<String, OrderEvent> medValue   = branches.get("tier-medium");
KStream<String, OrderEvent> lowValue   = branches.get("tier-low");

// Route each tier to a different sink topic
highValue.to("labs.orders.high-value");
medValue.to("labs.orders.medium-value");
lowValue.to("labs.orders.low-value");

Predicate evaluation is first-match. An order with amount=600 matches the first predicate (> 500) and goes to tier-high — it is never evaluated against > 100. Always order predicates from most to least specific.

Kafka 3 → Kafka 4 migration: KStream.branch(Predicate[]) returns an array and is deprecated. Switch to stream.split(Named.as("prefix-")).branch(...).defaultBranch(...) which returns a named Map.

5. peek — side-effects without transforming

Cardinality: 1 → 1 (passthrough). Records flow through unchanged — peek just lets you observe them.

KStream<String, OrderEvent> withLogging = orders
    .peek((key, value) ->
        log.info("Processing order key={} amount={}", key, value.getAmount()))
    .filter((k, v) -> v.getStatus() == Status.CONFIRMED);

Use peek() for logging, metrics emission, or debugging. Never use it for side-effects that must be exactly-once (use a proper sink for that).

6. Chaining operators

Operators chain fluently — each returns a new KStream:

KStream<String, OrderEvent> confirmed = builder
    .stream("labs.events")                                          // source
    .peek((k, v) -> log.debug("received {}", v.getOrderId()))      // log
    .filter((k, v) -> v.getStatus() == Status.CONFIRMED)           // filter
    .mapValues(v -> enrichWithUserData(v))                         // enrich (no rekey)
    .selectKey((k, v) -> v.getUserId().toString());                // rekey (repartition)

KTable<String, Long> countPerUser = confirmed
    .groupByKey()
    .count(Materialized.as("order-counts"));

countPerUser.toStream().to("labs.order-counts");                   // sink

7. Stateless operator quick-reference

8. The .repartition() operator — explicit control

Day 29 §9 covered that key-changing operators create an automatic, implicit repartition topic before the next stateful operation. Kafka Streams also offers .repartition() as an explicit operator — useful when you want direct control over the repartition topic’s name, partition count, or when you need to force a repartition point deliberately rather than relying on the DSL inferring one.

KStream<String, OrderEvent> repartitioned = confirmed
    .repartition(Repartitioned.<String, OrderEvent>as("orders-by-user")
        .withNumberOfPartitions(12)   // explicit control — doesn't have to match source topic
        .withKeySerde(Serdes.String())
        .withValueSerde(orderEventSerde));

Why use this instead of relying on the implicit repartition from selectKey+groupByKey:

• Naming control — the implicit repartition topic gets an auto-generated name ({application-id}-{internal-node-name}-repartition); .repartition() lets you give it a stable, predictable name that survives topology changes elsewhere in the app (auto-generated names can shift if you add/remove unrelated operators upstream, which silently orphans the old internal topic).
• Partition count control — the implicit repartition inherits the source topic’s partition count; .repartition() lets you deliberately size it differently (e.g. more partitions for a downstream aggregation that needs more parallelism than the source topic has).
• Reuse across multiple downstream branches — if several different aggregations need the same re-keyed stream, an explicit named repartition avoids Kafka Streams silently creating multiple equivalent internal topics for what’s conceptually the same repartitioning.

Rule of thumb: the implicit repartition is fine for a simple, one-off selectKey → groupByKey chain. Reach for explicit .repartition() once the topology has multiple branches, you care about topic naming stability across deploys, or you need a partition count that differs from the source topic.

9 Exceptions inside stateless operators — what actually happens

A filter/map/flatMap lambda that throws is not automatically caught and routed anywhere — by default, an uncaught exception kills the StreamThread processing that task, and depending on default.production.exception.handler/thread-level configuration, can bring down the whole Streams app instance if not handled.

// Dangerous — a malformed record or null-pointer here kills the processing thread
KStream<String, Double> amounts = orders
    .mapValues(value -> value.getAmount() / value.getQuantity()); // ArithmeticException if quantity=0

Production pattern — handle it explicitly inside the lambda, don’t rely on framework-level recovery for business logic errors:

KStream<String, Double> amounts = orders
    .mapValues(value -> {
        try {
            return value.getQuantity() == 0 ? 0.0 : value.getAmount() / value.getQuantity();
        } catch (Exception e) {
            log.error("Failed to compute unit price for order={}", value.getOrderId(), e);
            return 0.0; // or route to a dead-letter stream via branching (§4) before this point
        }
    });

# Framework-level safety net for DESERIALIZATION errors specifically (not business logic)
spring:
  kafka:
    streams:
      default-deserialization-exception-handler: LogAndContinueExceptionHandler
      # LogAndFailExceptionHandler is the default — LogAndContinue skips the bad record instead of crashing

The distinction that matters: default.deserialization.exception.handler only covers deserialization failures at the source — it does not protect against a NullPointerException or ArithmeticException thrown inside your own filter/map/flatMap lambda. Business logic exceptions inside DSL operators need their own try/catch, the same discipline as any other production code — Kafka Streams doesn’t give you Day 18’s DefaultErrorHandler-style DLT recovery for free inside stateless operators.

10. Testing filter, map, flatMap, and branch with TopologyTestDriver

Building on Day 29 §11’s introduction, here’s the pattern applied specifically to stateless operators — fast, no broker needed.

class OrderTopologyStatelessTest {

    TopologyTestDriver testDriver;
    TestInputTopic<String, OrderEvent> input;
    TestOutputTopic<String, OrderEvent> highValueOutput;

    @BeforeEach
    void setup() {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, OrderEvent> orders = builder.stream("labs.events");
        Map<String, KStream<String, OrderEvent>> branches = orders
            .split(Named.as("tier-"))
            .branch((k, v) -> v.getAmount() > 500, Branched.as("high"))
            .defaultBranch(Branched.as("low"));
        branches.get("tier-high").to("labs.orders.high-value");

        testDriver = new TopologyTestDriver(builder.build(), testProps());
        input = testDriver.createInputTopic("labs.events", stringSer, avroSer);
        highValueOutput = testDriver.createOutputTopic("labs.orders.high-value", stringDeser, avroDeser);
    }

    @Test
    void routesHighValueOrdersToCorrectBranch() {
        input.pipeInput("k1", orderWithAmount(600.0));
        input.pipeInput("k2", orderWithAmount(50.0));  // should NOT appear in high-value output

        assertThat(highValueOutput.readValue().getAmount()).isEqualTo(600.0);
        assertThat(highValueOutput.isEmpty()).isTrue(); // confirms the low-value order was correctly excluded
    }

    @Test
    void firstMatchingBranchWinsWhenPredicatesOverlap() {
        // Regression test for the exact footgun in §4 — an amount matching
        // multiple predicates must only be routed to the FIRST match
        input.pipeInput("k1", orderWithAmount(600.0)); // matches both >500 and >100
        assertThat(highValueOutput.readValue()).isNotNull(); // went to "high", not "medium"
    }
}

The second test matters more than it looks — branch predicate ordering bugs (§4’s warning) are exactly the kind of mistake that’s easy to introduce during a later refactor (reordering .branch() calls) and won’t show up as a compile error or an obvious runtime failure, just silently wrong routing. A dedicated regression test for overlapping-predicate behavior catches it immediately.

11. Monitoring repartition topic cost

Since both implicit (Day 29 §9) and explicit (§8) repartitioning create real Kafka topics, they’re worth watching like any other topic — easy to forget because they don’t appear anywhere in your own topic-creation code.

# List all internal topics for this Streams app — repartition + changelog topics
# both follow the {application-id}-... naming convention
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep labs-streams-app

Practical habit: after any topology change involving map/selectKey/flatMap before a stateful operation, re-run topology.describe() and diff the internal topic list against what you expect — catching an accidental extra repartition topic in review is much cheaper than discovering it via a surprise disk usage alert later.

12. Common pitfalls

• Chaining multiple key-changing operations before a single groupByKey — each key change before the eventual stateful operation can create its own repartition topic if not consolidated; prefer a single selectKey/map immediately before the stateful op, or use explicit .repartition() (§8) to make the intent and cost visible
• Assuming default.deserialization.exception.handler covers business logic exceptions — it only catches deserialization failures, not exceptions thrown inside your own lambda bodies (§9)
• Ordering branch() predicates from least to most specific — silently misroutes records that match multiple predicates to the wrong branch, since only the first match wins (§4)
• Using peek() for anything that needs exactly-once delivery guarantees — it’s explicitly a side-effect/observation hook, not a sink with delivery guarantees
• Not testing overlapping-predicate branch behavior — a routing bug introduced by reordering .branch() calls during a refactor won’t surface as a compile error or obvious crash (§10)

Key Takeaways

• filter: 1-to-0/1 — drop records; filterNot is the inverse predicate shorthand
• mapValues: 1-to-1 no repartition ✅ — prefer over map() when key doesn’t change
• map / selectKey: changing the key triggers an internal repartition topic (extra latency)
• flatMap: 1-to-N — explode one record into many (e.g. order → individual line items)
• split().branch(): Kafka 4 API routes each record to exactly one named sub-stream — first match wins, order predicates from most to least specific
• .repartition() gives explicit control over the internal topic’s name and partition count — reach for it once a topology has multiple branches or naming-stability needs
• Uncaught exceptions inside filter/map/flatMap lambdas kill the processing thread — default.deserialization.exception.handler does NOT cover this, only actual deserialization failures
• TopologyTestDriver tests stateless operators fast and without a broker — always include a regression test for overlapping branch predicates
• All operators are lazy — nothing executes until the topology is started by Kafka Streams

---

Support me through GitHub Sponsors.

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

Next

➡️ Day 31: Windowing — tumbling, hopping, session windows

Resources

• 📘 Kafka: The Definitive Guide — Chapter 11 (Kafka Streams)
• 🌐 kafka.apache.org/documentation/streams/developer-guide/dsl-api
• 🌐 Kafka Streams — repartition() operator

👉 Link to Medium blog