60-Day Kafka 4 Learning Plan · Week 5 — Day 34 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/interactive-queries

Goal

Understand how Kafka Streams persists stateful data locally using RocksDB, how the changelog topic provides fault tolerance, how to expose state store contents through interactive REST queries — including multi-instance routing, standby replicas, and what happens to queries during a rebalance — and the production RocksDB tuning knobs worth knowing.

1. What is a state store?

A state store is a local database embedded in each Streams instance used for stateful operations: aggregations, joins, and custom processors. Each store is:

• Local — data lives on the same JVM that processes the partition
• Backed by a changelog — every write is mirrored to a compacted Kafka topic
• Fault-tolerant — on restart, the store is rebuilt by replaying its changelog topic
• Queryable — reads are sub-millisecond, directly from local disk (no Kafka round-trip)

Three store types

2. RocksDB + changelog lifecycle

KStream event (labs.events)
  → aggregate() processor
  → RocksDB state store (stats-store, local disk)
  → changelog topic (stats-store-changelog, compacted Kafka topic)

On restart:
  changelog topic → replay → rebuild RocksDB → ready to serve

RocksDB is a Log-Structured Merge-tree (LSM) database optimised for write-heavy workloads:

• Writes go to an in-memory MemTable first (very fast)
• Periodically flushed to immutable SSTable files on disk
• Background compaction merges SSTables and removes old values

The changelog topic uses log compaction — Kafka retains only the latest value per key, so replay is efficient even after long periods.

3. Persistent vs in-memory stores

Choosing between persistent and in-memory

// Persistent (default) — survives restarts
Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("count-store")
    .withKeySerde(Serdes.String())
    .withValueSerde(Serdes.Long());

// In-memory — fast, no disk writes, lost on restart
Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("count-store")
    .withStoreType(Materialized.StoreType.IN_MEMORY);

Use in-memory stores only for:

• Unit tests (fast, no cleanup needed)
• Truly ephemeral state where loss on restart is acceptable
• Very small datasets that fit entirely in JVM heap

4. Creating stores — Materialized API

// Persistent KeyValue store with full Serde config
KTable<String, OrderStats> stats = orders
    .groupByKey()
    .aggregate(
        () -> new OrderStats(0, 0.0),
        (key, event, acc) -> new OrderStats(acc.count() + 1, acc.sum() + event.getAmount()),
        Materialized.<String, OrderStats, KeyValueStore<Bytes, byte[]>>
            as("stats-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(orderStatsSerde)
            .withRetention(Duration.ofDays(7))      // keep changelog for 7 days
            .withCachingEnabled()                    // batch writes (reduces changelog volume)
    );

// Windowed store with explicit retention
KTable<Windowed<String>, Long> windowedCounts = orders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count(
        Materialized.<String, Long, WindowStore<Bytes, byte[]>>
            as("windowed-count-store")
            .withRetention(Duration.ofHours(1))     // keep 1 hour of windows
    );

Caching

withCachingEnabled() (default) batches writes before flushing to RocksDB and the changelog. This reduces downstream KTable update frequency — useful for high-throughput aggregations. Disable caching with withCachingDisabled() when you need every intermediate update downstream.

5. Interactive queries — query store directly

Read from the state store inside a REST endpoint — no Kafka consumer, no extra hop:

@RestController
@RequiredArgsConstructor
public class OrderStatsController {

    private final KafkaStreams streams;

    // Single key lookup
    @GetMapping("/stats/{userId}")
    public OrderStats getStats(@PathVariable String userId) {
        ReadOnlyKeyValueStore<String, OrderStats> store = streams.store(
            StoreQueryParameters.fromNameAndType(
                "stats-store",
                QueryableStoreTypes.keyValueStore()
            )
        );
        OrderStats result = store.get(userId);
        if (result == null) throw new ResponseStatusException(HttpStatus.NOT_FOUND);
        return result;
    }

    // Range scan — all users with IDs between "a" and "m"
    @GetMapping("/stats/range")
    public List<OrderStats> getRange() {
        ReadOnlyKeyValueStore<String, OrderStats> store = streams.store(
            StoreQueryParameters.fromNameAndType("stats-store",
                QueryableStoreTypes.keyValueStore())
        );
        List<OrderStats> result = new ArrayList<>();
        store.range("a", "m").forEachRemaining(kv -> result.add(kv.value));
        return result;
    }

    // Windowed store query — fetch count for user in last 5 minutes
    @GetMapping("/windowed/{userId}")
    public long getWindowedCount(@PathVariable String userId) {
        ReadOnlyWindowStore<String, Long> windowStore = streams.store(
            StoreQueryParameters.fromNameAndType(
                "windowed-count-store",
                QueryableStoreTypes.windowStore()
            )
        );
        Instant now = Instant.now();
        WindowStoreIterator<Long> it = windowStore.fetch(userId,
            now.minus(5, ChronoUnit.MINUTES), now);
        return it.hasNext() ? it.next().value : 0L;
    }
}

6. Distributed state — multi-instance routing

When multiple Streams instances process different partitions, each instance only holds its assigned partition slice. A request for key user-42 must go to the instance that owns partition hash("user-42") % numPartitions.

@GetMapping("/stats/{userId}")
public OrderStats getStats(@PathVariable String userId) {

    // 1. Find which instance owns this key's partition
    KeyQueryMetadata meta = streams.queryMetadataForKey(
        "stats-store",
        userId,
        Serdes.String().serializer()
    );

    HostInfo owner = meta.activeHost();

    // 2. Local read if this instance owns it
    if (owner.host().equals(myHostname) && owner.port() == myPort) {
        ReadOnlyKeyValueStore<String, OrderStats> store = streams.store(
            StoreQueryParameters.fromNameAndType("stats-store",
                QueryableStoreTypes.keyValueStore())
        );
        return store.get(userId);
    }

    // 3. Proxy to the owning instance via HTTP
    String url = "http://" + owner.host() + ":" + owner.port() + "/stats/" + userId;
    return restTemplate.getForObject(url, OrderStats.class);
}

application.yml — advertise this instance’s host

spring:
  kafka:
    streams:
      application-id: labs-streams-app
      bootstrap-servers: localhost:9092
      properties:
        application.server: labs-api-1:8080   # advertised host:port for inter-instance routing

7. Standby replicas — trading disk/network for read availability

Day 29 §8 introduced num.standby.replicas as a way to speed up rebalance recovery. For interactive queries specifically, standbys also unlock stale-but-available reads during a failover, if you choose to use them.

spring:
  kafka:
    streams:
      properties:
        num.standby.replicas: 1

meta.standbyHosts() (alongside meta.activeHost() from §6) returns instances holding a standby copy of the partition — not the active/authoritative one, but a continuously-updated replica that’s typically only moments behind.

// Optional: fall back to a standby if the active host is temporarily unreachable
// (e.g. mid-rebalance — see §9) rather than failing the request outright
Set<HostInfo> standbys = meta.standbyHosts();
if (activeHostUnreachable && !standbys.isEmpty()) {
    HostInfo standby = standbys.iterator().next();
    // Query the standby instead — the caller must accept it may be a few records behind
}

The trade-off to be explicit about: querying a standby means accepting eventual consistency for that read — the standby’s local RocksDB lags the active copy by however long it takes changelog updates to propagate, typically small but not zero. This is a deliberate choice between “always exact, sometimes momentarily unavailable during rebalance” (active-only) and “always available, occasionally slightly stale” (active-with-standby-fallback) — pick based on whether your use case (§9 discusses when this actually comes up) can tolerate the staleness.

8. RocksDB tuning — the knobs worth knowing about

The default RocksDB configuration is reasonable for moderate workloads, but two settings matter enough to know explicitly once an aggregation’s key cardinality or write rate grows:

public class CustomRocksDBConfig implements RocksDBConfigSetter {
    @Override
    public void setConfig(String storeName, Options options, Map<String, Object> configs) {
        BlockBasedTableConfig tableConfig = new BlockBasedTableConfig();
        tableConfig.setBlockCache(new LRUCache(64 * 1024 * 1024)); // 64MB block cache — bigger = fewer disk reads for hot keys
        options.setTableFormatConfig(tableConfig);

        options.setWriteBufferSize(32 * 1024 * 1024);   // 32MB memtable — bigger = fewer, larger flushes
        options.setMaxWriteBufferNumber(3);
    }
}

spring:
  kafka:
    streams:
      properties:
        rocksdb.config.setter: com.labs.streams.CustomRocksDBConfig

Default sizing multiplies per state store, not per app. Each named Materialized.as(...) store gets its own RocksDB instance with its own memtable/cache allocation — an app with several large aggregations can accumulate significant memory overhead from defaults alone, well before any deliberate tuning. Account for num_stores × per_store_memory when sizing instance memory, not just a single store’s footprint.

9. Querying during a rebalance — the availability gap

Standby replicas (§7) exist partly to soften this: during a rebalance (an instance joins/leaves, Day 6 §4’s mechanics apply directly since a Streams app is a consumer group), a partition — and its state store — is mid-transfer between instances for a brief window. A query for a key in that partition can find no instance currently reporting itself as the active owner.

Rebalance starts → partition 3 revoked from instance A
                 → (brief gap — no instance is the confirmed active owner yet)
                 → partition 3 assigned to instance B
                 → instance B replays/catches up on its state store
                 → instance B becomes active owner

Handling this in the REST layer:

@GetMapping("/stats/{userId}")
public ResponseEntity<OrderStats> getStats(@PathVariable String userId) {
    try {
        KeyQueryMetadata meta = streams.queryMetadataForKey("stats-store", userId, keySerializer);
        if (meta == KeyQueryMetadata.NOT_AVAILABLE) {
            // Rebalance in progress — no confirmed owner right now
            return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE)
                .header("Retry-After", "2")
                .build();
        }
        // ... proceed with normal routing from §6, or fall back to a standby per §7
    } catch (InvalidStateStoreException e) {
        // This instance's local store isn't ready yet either — same signal, different exception path
        return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build();
    }
}

Design the client (or an API gateway retry policy) to expect this, not treat it as an error. A rebalance-induced SERVICE_UNAVAILABLE is transient and self-resolving within seconds under normal conditions — the right response is a short retry with backoff, not surfacing a hard failure to an end user. This is conceptually the same “don’t treat routine coordination as an error” framing from Day 27 §7’s discussion of merge conflicts — brief unavailability during a rebalance is expected operational behavior, not a bug.

10. Testing multi-instance routing behavior

Full multi-instance routing is hard to unit test meaningfully (it requires actual multiple JVMs with real partition assignment), but the routing decision logic can and should be tested in isolation.

@Test
void routesToLocalStoreWhenThisInstanceOwnsTheKey() {
    KeyQueryMetadata meta = mock(KeyQueryMetadata.class);
    when(meta.activeHost()).thenReturn(new HostInfo("this-instance", 8080));
    when(streamsMock.queryMetadataForKey(any(), any(), any(StringSerializer.class))).thenReturn(meta);

    // Assert the controller takes the LOCAL path, never calls restTemplate
    controller.getStats("u1");
    verifyNoInteractions(restTemplate);
}

@Test
void proxiesToRemoteInstanceWhenAnotherInstanceOwnsTheKey() {
    KeyQueryMetadata meta = mock(KeyQueryMetadata.class);
    when(meta.activeHost()).thenReturn(new HostInfo("other-instance", 8080));
    when(streamsMock.queryMetadataForKey(any(), any(), any(StringSerializer.class))).thenReturn(meta);

    controller.getStats("u1");
    verify(restTemplate).getForObject(contains("other-instance"), eq(OrderStats.class));
}

What this actually validates: not that distributed routing “works” end-to-end (that needs a real multi-instance integration environment, closer to a staging deployment than a unit test), but that the routing decision — local vs proxy — is made correctly given a known ownership answer. That’s the part most likely to have a bug (an off-by-one host/port comparison, wrong exception handling) and the part cheapest to test in isolation.

11. Common pitfalls

• Never accounting for num_stores × per_store_memory — RocksDB defaults apply per named store, not per app; several large aggregations can add up to significant memory well before any deliberate tuning happens (§8)
• Treating a rebalance-time SERVICE_UNAVAILABLE as a hard failure instead of a transient, retry-appropriate condition — surfaces routine coordination as user-facing errors unnecessarily (§9)
• Querying standbys without accepting the staleness trade-off explicitly — silently returning slightly-stale data as if it were authoritative can surprise callers who assumed interactive queries are always exactly current (§7)
• Only testing the “happy path” local-read case — the proxy-to-remote-instance path and the rebalance-unavailable path are both easy to leave untested despite being common in any real multi-instance deployment (§10)
• Assuming withCachingEnabled()‘s batching means data loss risk — it doesn’t; caching only affects how often downstream updates are emitted, not durability (writes still go to RocksDB and the changelog) — don’t disable it reflexively without an actual need for every intermediate update

Key Takeaways

• Three store types: KeyValue (aggregations), Window (windowed), Session (session windows)
• RocksDB is the default persistent backend — LSM tree, fast writes, compacted reads
• Every write is also written to a changelog Kafka topic — survives restarts via replay
• Interactive queries: streams.store() reads RocksDB directly — no Kafka round-trip needed
• Multi-instance: use queryMetadataForKey() to find which instance owns the partition
• Standby replicas enable stale-but-available reads during rebalance — a deliberate consistency/availability trade-off, not a free upgrade
• RocksDB tuning (block cache, write buffer size) matters once key cardinality or write rate grows — and multiplies per named state store, not per app
• Rebalance creates a brief window where no instance is a confirmed active owner — design for SERVICE_UNAVAILABLE as expected, retry-appropriate behavior
• In-memory stores: fast but no persistence — for tests and short-lived state only
• withCachingEnabled() (default) batches updates for emission frequency, not durability — disable with withCachingDisabled() only when every intermediate result matters downstream

---

Support me through GitHub Sponsors.

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

Next

➡️ Day 35: Streams project — real-time analytics with Spring Cloud Stream

Resources

• 📘 Kafka: The Definitive Guide — Chapter 11 (State Stores)
• 🌐 kafka.apache.org — Interactive Queries
• 🌐 kafka.apache.org — State Stores
• 🌐 RocksDB — Tuning Guide

Link to Medium blog