Metis

Field note / kafka

publishedupdated 2026-05-01#kafka#streams#java#eda

Kafka Streams: Stateful Processing

Building idempotent, exactly-once data pipelines with Kafka Streams — topology design, state stores, and windowed aggregations.

Overview

Kafka Streams is a client library for stream processing — your topology runs inside your application process with no external cluster needed. It provides:

  • Stateful operations via local RocksDB state stores (backed by Kafka changelog topics)
  • Exactly-once processing guarantees (processing.guarantee=exactly_once_v2)
  • Native support for joins, aggregations, and windowed operations
  • Built-in metrics via JMX

Kafka Streams replaces fragile consumer-producer loops with declarative, testable topologies that can scale across partitions while keeping state local to the application.

When to use Kafka Streams:

  • Your state is small-to-medium (< 100GB per partition)
  • You need joins between streams/tables without an external DB hop
  • You want testable stream logic (TopologyTestDriver)
  • You're already on the JVM

When NOT to use it:

  • Very large state (> 100GB) → prefer Apache Flink
  • Sub-10ms latency SLA → overhead of exactly-once is too high
  • Non-JVM stack → use a native Kafka client library

Technical Implementation

Topology Design

StreamsBuilder builder = new StreamsBuilder();

// Source stream — raw item events
KStream<String, ItemEvent> items = builder.stream(
    "item-events",
    Consumed.with(Serdes.String(), itemSerde)
);

// Reference table — supplier metadata (compacted topic, changelog-backed)
KTable<String, SupplierData> suppliers = builder.table(
    "supplier-data",
    Materialized.<String, SupplierData, KeyValueStore<Bytes, byte[]>>as("supplier-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(supplierSerde)
);

// Enrichment join: stream-table (never triggers on table update)
KStream<String, EnrichedItem> enriched = items.join(
    suppliers,
    (item, supplier) -> EnrichedItem.builder()
        .itemId(item.getId())
        .supplierId(supplier.getId())
        .price(item.getPrice())
        .supplierRegion(supplier.getRegion())
        .build(),
    Joined.with(Serdes.String(), itemSerde, supplierSerde)
);

// Tumbling window aggregation — 5-minute rollup per category
enriched
    .groupBy((k, v) -> v.getCategory(), Grouped.with(Serdes.String(), enrichedSerde))
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .aggregate(
        CategoryRollup::empty,
        (key, value, agg) -> agg.add(value),
        Materialized.<String, CategoryRollup, WindowStore<Bytes, byte[]>>as("category-rollup-store")
            .withKeySerde(Serdes.String())
            .withValueSerde(rollupSerde)
    )
    .toStream()
    .map((windowedKey, rollup) -> KeyValue.pair(windowedKey.key(), rollup))
    .to("category-rollups", Produced.with(Serdes.String(), rollupSerde));

Exactly-Once Configuration

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "item-enrichment-pipeline");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);
// Commit interval — lower = less re-processing on restart, higher = better throughput
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);

EXACTLY_ONCE_V2 uses cooperative rebalancing and transactional producers. Throughput cost vs at-least-once: ~15–20%. Worth it for inventory data where duplicates cause double-counting in reports.

RocksDB Tuning for High Volume

Default RocksDB settings can underperform at sustained high event rates. A production configuration should tune:

public class HighVolumeRocksDBConfig implements RocksDBConfigSetter {
    @Override
    public void setConfig(String storeName, Options options, Map<String, Object> configs) {
        options.setWriteBufferSize(64 * 1024 * 1024L);      // 64MB write buffer
        options.setMaxWriteBufferNumber(3);
        options.setTargetFileSizeBase(64 * 1024 * 1024L);   // 64MB SST files
        options.setCompressionType(CompressionType.LZ4_COMPRESSION);
        options.setMaxOpenFiles(100);                         // prevent fd exhaustion
    }
}

Testing with TopologyTestDriver

@Test
void enrichedItemJoinsSupplierData() {
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());

    try (TopologyTestDriver driver = new TopologyTestDriver(buildTopology(), props)) {
        TestInputTopic<String, SupplierData> supplierTopic = driver.createInputTopic(
            "supplier-data", Serdes.String().serializer(), supplierSerde.serializer()
        );
        TestInputTopic<String, ItemEvent> itemTopic = driver.createInputTopic(
            "item-events", Serdes.String().serializer(), itemSerde.serializer()
        );
        TestOutputTopic<String, EnrichedItem> outputTopic = driver.createOutputTopic(
            "enriched-items", Serdes.String().deserializer(), enrichedSerde.deserializer()
        );

        supplierTopic.pipeInput("SUP-1", new SupplierData("SUP-1", "West"));
        itemTopic.pipeInput("ITEM-1", new ItemEvent("ITEM-1", "SUP-1", 29.99));

        EnrichedItem result = outputTopic.readValue();
        assertEquals("West", result.getSupplierRegion());
        assertEquals(29.99, result.getPrice());
    }
}

Architecture

Item enrichment pipeline: stream-table join feeds a windowed category rollup

State store topology:

  • supplier-store — KTable backed by RocksDB, replicated via changelog topic
  • category-rollup-store — windowed store, 5-min retention
  • Both stores are partition-local — each Streams task owns its partition's state

Challenges

1. Rebalance storms under rolling deploys

With EXACTLY_ONCE_V2, rebalancing pauses processing. During a rolling deploy with 4 app instances, we saw cascading rebalances — each new instance joining triggered a full group rebalance, causing 30–60s processing gaps.

Fix: switched to sticky partition assignment + increased session.timeout.ms to 60s.

props.put(StreamsConfig.RACK_AWARE_ASSIGNMENT_TAGS_CONFIG, "az");
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 60_000);

2. State store size blowup

After 3 months in production, supplier-store RocksDB size on disk grew to 80GB per partition due to a changelog compaction misconfiguration. The topic wasn't being compacted aggressively enough.

Fix: enforced segment.bytes=67108864 and min.cleanable.dirty.ratio=0.1 on the changelog topic.

3. Dead-letter topic (retrofitted)

We didn't add DLT routing initially. When malformed events arrived (supplier JSON schema drift), the Streams app threw deserialization exceptions and crashed. We added a custom DeserializationExceptionHandler that routes bad records to a item-events-dlq topic.

props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
    DLQDeserializationExceptionHandler.class);

Production Checklist

  • Design the topology around explicit keys, partitioning, and replay behavior.
  • Tune RocksDB and capacity-plan state stores against realistic event rates.
  • Roll out EXACTLY_ONCE_V2 gradually and measure its rebalance and latency impact.
  • Build deterministic tests with TopologyTestDriver before integration testing.
  • Operate the pipeline with lag, rebalance, state-store, and restoration metrics.

References