60-Day Kafka 4 Learning Plan · Week 5 — Day 32 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#aggregating

Goal

Master Kafka Streams stateful aggregations — count, reduce, and the fully custom aggregate — understand how RocksDB state stores work under the hood, query them interactively from a REST endpoint, and fix the gap in that setup once the app scales to multiple instances.

1 Group before you aggregate

Every aggregation requires a grouping step first.

// groupByKey() ★ — uses existing key, NO repartition
KGroupedStream<String, OrderEvent> grouped = orders.groupByKey();

// groupBy(selector) ⚠ — derives new key, TRIGGERS repartition
KGroupedStream<String, OrderEvent> byStatus = orders
    .groupBy((key, value) -> value.getStatus().toString());

Always prefer groupByKey() when the key is already correct — groupBy() creates an internal repartition topic that adds latency and storage overhead.

2. count — events per key

Counts the number of records per grouped key. Simplest aggregation — produces a KTable<K, Long>.

// Count confirmed orders per userId
KTable<String, Long> orderCount = orders
    .filter((k, v) -> v.getStatus() == Status.CONFIRMED)
    .groupByKey()
    .count(Materialized.as("order-count-store"));  // named RocksDB state store

// Emit count updates to output topic
orderCount.toStream().to("labs.order-counts",
    Produced.with(Serdes.String(), Serdes.Long()));

The Materialized.as("store-name") gives the backing state store a name — required for interactive queries (§5).

3 reduce — combine values of the same type

Combines consecutive values using a binary function. The input value type and the output type must be the same.

// Running total spend per userId — Double + Double → Double
KStream<String, Double> amounts = orders
    .mapValues(v -> v.getAmount());  // extract Double, key unchanged

KTable<String, Double> totalSpend = amounts
    .groupByKey()
    .reduce(
        (runningTotal, newAmount) -> runningTotal + newAmount,
        Materialized.as("total-spend-store")
    );

// Max order value per userId
KTable<String, Double> maxOrder = amounts
    .groupByKey()
    .reduce(Math::max, Materialized.as("max-order-store"));

Limitation: When the first record arrives, it becomes the initial value (no separate initializer). If you need a custom starting value, use aggregate().

4 aggregate — full custom accumulator

The most powerful operator. Define an initializer (starting state) and an aggregator (how to fold each record into the accumulator). Output can be any type — different from the input.

// Accumulator: track count + sum to derive average order value
record OrderStats(long count, double sum) {
    double avg() { return count == 0 ? 0.0 : sum / count; }
}

KTable<String, OrderStats> stats = orders
    .groupByKey()
    .aggregate(
        () -> new OrderStats(0, 0.0),           // initializer — called once per key
        (userId, event, acc) -> new OrderStats( // aggregator — called per record
            acc.count() + 1,
            acc.sum() + event.getAmount()
        ),
        Materialized.<String, OrderStats, KeyValueStore<Bytes, byte[]>>
            as("stats-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(/* custom Serde for OrderStats */)
    );

// Derive average as a separate stream
stats.toStream()
     .mapValues(s -> s.avg())
     .to("labs.avg-order-value");

Custom Serde for OrderStats (JSON example)

Serde<OrderStats> statsSerde = Serdes.serdeFrom(
    (topic, data) -> objectMapper.writeValueAsBytes(data),
    (topic, bytes) -> objectMapper.readValue(bytes, OrderStats.class)
);

Handle null explicitly in a custom Serde. Kafka Streams calls the serializer/deserializer with null in legitimate cases (e.g. tombstones, or a key that’s never been seen by get()-style lookups) — a naive objectMapper.writeValueAsBytes(data) throws a NullPointerException on null input instead of returning null bytes, which crashes the StreamThread exactly like the uncaught-exception scenario from Day 30 §9. Guard both directions: data == null ? null : objectMapper.writeValueAsBytes(data).

5 State stores — where aggregations live

Every stateful aggregation materializes into a RocksDB state store on local disk, backed by a compacted changelog Kafka topic.

KStream (labs.events)
  → groupByKey().aggregate()
  → RocksDB state store (stats-store)     ← local on-disk, fast reads
  → backed by changelog topic              ← compacted Kafka topic, replays on restart

Fault tolerance: If a Streams instance restarts, it replays the changelog topic to rebuild the local RocksDB state — no data loss.

Interactive queries — read state store directly

@RestController
@RequiredArgsConstructor
public class StatsController {

    private final KafkaStreams streams;

    @GetMapping("/stats/{userId}")
    public OrderStats getStats(@PathVariable String userId) {
        ReadOnlyKeyValueStore<String, OrderStats> store = streams.store(
            StoreQueryParameters.fromNameAndType(
                "stats-store",
                QueryableStoreTypes.keyValueStore()
            )
        );
        return store.get(userId);  // reads local RocksDB — no Kafka round-trip
    }

    @GetMapping("/stats")
    public Map<String, OrderStats> getAllStats() {
        ReadOnlyKeyValueStore<String, OrderStats> store = streams.store(
            StoreQueryParameters.fromNameAndType("stats-store",
                QueryableStoreTypes.keyValueStore())
        );
        Map<String, OrderStats> result = new HashMap<>();
        store.all().forEachRemaining(kv -> result.put(kv.key, kv.value));
        return result;
    }
}

Windowed aggregations with state stores

// Count orders per userId per 5-min tumbling window — windowed state store
KTable<Windowed<String>, Long> windowedCounts = orders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count(Materialized.as("windowed-count-store"));

// Read windowed store
ReadOnlyWindowStore<String, Long> windowStore = streams.store(
    StoreQueryParameters.fromNameAndType(
        "windowed-count-store",
        QueryableStoreTypes.windowStore()
    )
);
long count = windowStore.fetch("user-123", Instant.now().minus(5, MINUTES), Instant.now());

6 count vs reduce vs aggregate

7 The multi-instance interactive query problem — this lab’s biggest hidden gap

The StatsController in §5 works perfectly with one Streams app instance. It silently returns wrong results (nulls for keys that definitely exist) the moment you run more than one — the exact same class of bug as the WebSocket multi-instance gap from Day 20 §7, just for state stores instead of WebSocket sessions.

Why: a state store’s data is partitioned across all running instances of the same application-id — each instance only holds the RocksDB shard for the partitions assigned to it (Day 29 §10). streams.store(...) only ever queries this instance’s local shard. If userId="u1"‘s data lives on partition 3, and partition 3 is currently assigned to instance B, then calling GET /stats/u1 against instance A returns null — not because the data doesn’t exist, but because A is looking in the wrong place.

Client → Load balancer → instance A (queries ITS local RocksDB shard only)

But userId="u1" hashes to partition 3, which is assigned to instance B

instance A's local store has no entry for "u1" → returns null,
even though the real answer exists two machines away

The fix — route to the correct instance using Streams’ metadata API:

@GetMapping("/stats/{userId}")
public ResponseEntity<OrderStats> getStats(@PathVariable String userId) {
    KeyQueryMetadata metadata = streams.queryMetadataForKey(
        "stats-store", userId, new StringSerializer());

    HostInfo activeHost = metadata.activeHost();
    if (isThisInstance(activeHost)) {
        // The key IS on this instance — query locally, no network hop
        ReadOnlyKeyValueStore<String, OrderStats> store = streams.store(
            StoreQueryParameters.fromNameAndType("stats-store", QueryableStoreTypes.keyValueStore()));
        return ResponseEntity.ok(store.get(userId));
    } else {
        // The key is on a DIFFERENT instance — forward the request there
        String remoteUrl = String.format("http://%s:%d/stats/%s",
            activeHost.host(), activeHost.port(), userId);
        return restTemplate.getForEntity(remoteUrl, OrderStats.class);
    }
}

# Each instance must advertise where it can be reached for this RPC forwarding to work
spring:
  kafka:
    streams:
      properties:
        application.server: ${HOSTNAME}:8080   # this instance's own reachable address

This is not optional for any production deployment expecting more than one instance. Exactly like Day 20’s WebSocket gap, this fails silently rather than loudly — the demo works perfectly with one instance, and the first symptom in production is “some users’ stats randomly return null,” which is a much harder bug to diagnose after the fact than designing for it up front.

8 Testing aggregations with TopologyTestDriver

Building on the pattern from Day 30 §10, testing aggregation logic means asserting against the state store directly, not just the output topic.

@Test
void aggregatesRunningStatsPerUser() {
    input.pipeInput("u1", orderWithAmount(100.0));
    input.pipeInput("u1", orderWithAmount(50.0));

    KeyValueStore<String, OrderStats> store = testDriver.getKeyValueStore("stats-store");
    OrderStats stats = store.get("u1");

    assertThat(stats.count()).isEqualTo(2);
    assertThat(stats.avg()).isEqualTo(75.0);
}

@Test
void handlesNullGracefullyInCustomSerde() {
    // Regression test for the §4 Serde null-handling warning —
    // a tombstone or absent-key lookup must not throw
    KeyValueStore<String, OrderStats> store = testDriver.getKeyValueStore("stats-store");
    assertThat(store.get("never-seen-user")).isNull(); // should return null cleanly, not throw
}

Querying the state store directly in tests (testDriver.getKeyValueStore(...)) is the aggregation-specific extension of the general TopologyTestDriver pattern — it lets you verify accumulator logic precisely, independent of whatever you eventually do with the output stream.

9 Monitoring state store health

Extending Day 29 §12’s Streams monitoring specifically for aggregation state stores:

# Changelog topics follow {application-id}-{store-name}-changelog naming
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep stats-store

10 Common pitfalls

Deploying the naive StatsController from §5 with more than one instance — the single biggest gap in this material; silently returns null for keys that exist on a different instance (§7)
Custom Serde that doesn’t handle null input — crashes the StreamThread on tombstones or certain lookup patterns, not just an edge case to skip (§4)
Forgetting to set application.server — without it, queryMetadataForKey can’t tell the caller where to forward a remote request, breaking the fix in §7 even if the routing code is otherwise correct
Testing only the output topic, never the state store directly — misses accumulator logic bugs that only show up in the store’s actual current value, not just what’s been emitted downstream (§8)
Using groupBy(selector) when the key was already correct — an unnecessary repartition topic (Day 30 §8/§9 cost) for no benefit over groupByKey()

Key Takeaways

groupByKey() keeps existing key (no repartition ✅); groupBy() derives new key (repartition ⚠️)
count: simplest — Long per key; reduce: same-type fold; aggregate: any-type accumulator
All aggregations are backed by a named RocksDB state store on local disk
State store is fault-tolerant via a changelog Kafka topic (compacted) — replays on restart
Custom Serdes must handle null explicitly — Kafka Streams passes it in legitimate cases (tombstones, misses)
Interactive queries against streams.store() only see this instance’s local shard — querying a key on a different instance silently returns null unless you route via queryMetadataForKey + application.server
Test aggregation logic by querying the state store directly in TopologyTestDriver, not just the output topic
Combine with windowedBy() (Day 31) for time-bounded aggregations per window

Support me through GitHub Sponsors.

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

➡️ Day 33: Stream joins — KStream-KTable & GlobalKTable

Resources

📘 Kafka: The Definitive Guide — Chapter 11 (Stateful Processing)
🌐 kafka.apache.org — Aggregating DSL
🌐 kafka.apache.org — Interactive Queries

👉 Link to Medium blog