Cheat SheetsApache KafkaKafka Streams

Kafka Streams — Cheat Sheet

Apache Kafka · 8 topics. Download the PDF or the Instagram carousel and share it.

Cheat Sheet · AiCanCode.org
Kafka Streams
Apache Kafka8 topicsQuick revision reference
1

Kafka Streams Basics

Kafka Streams is a Java client library for building stateful, fault-tolerant stream processing applications that read from and write to Kafka topics.

  • ✓Kafka Streams is an embedded Java library — no separate processing cluster required
  • ✓A topology is a DAG of source, processor, and sink nodes compiled from the Streams DSL
  • ✓Stateful operations use RocksDB state stores backed by compacted changelog topics
  • ✓On restart, state is restored from the changelog before processing resumes
  • ✓Scaling: run multiple instances of the same application.id; Kafka redistributes partitions
  • ✓Interactive Queries expose state store contents directly without an external database
Spring Boot — Kafka Streams topology with StreamsBuilder
@Configuration
@EnableKafkaStreams
class StreamConfig {
    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    KafkaStreamsConfiguration streamsConfig() {
        return new KafkaStreamsConfiguration(Map.of(
            StreamsConfig.APPLICATION_ID_CONFIG,    "order-processor",
            StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092",
            StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,   Serdes.String().getClass(),
            StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()
        ));
    }
}

@Component
class OrderTopology {
    @Autowired
    void buildPipeline(StreamsBuilder builder) {
        KStream<String, String> orders = builder.stream("orders-raw");

        orders
            .filter((key, value) -> value.contains(""status":"PAID""))
            .mapValues(value -> enrichOrder(value))
            .to("orders-enriched");   // sink topic
    }
}
2

KStream & KTable

KStream represents an unbounded event log; KTable represents a changelog-backed materialised view (latest value per key); GlobalKTable replicates fully to every instance.

  • ✓KStream = append-only event log; every record is independent. KTable = changelog; only the latest value per key matters.
  • ✓KStream-KTable joins look up the current table snapshot for each arriving stream event — useful for enrichment.
  • ✓KStream-KStream joins require a time window because both sides are unbounded streams.
  • ✓KTable state is backed by RocksDB locally and a changelog Kafka topic for fault tolerance and replication.
  • ✓GlobalKTable replicates fully to every instance — safe for small lookup tables; avoid for large datasets.
  • ✓Co-partitioning is required for KStream-KTable and KStream-KStream joins (same partition count, same partitioner).
Java — KStream filter, map, branch
StreamsBuilder builder = new StreamsBuilder();

// Source: read raw order events
KStream<String, OrderEvent> orders =
    builder.stream("order-events",
        Consumed.with(Serdes.String(), orderEventSerde));

// Stateless transforms
KStream<String, OrderEvent> validOrders = orders
    .filter((key, order) -> order.getAmount().compareTo(BigDecimal.ZERO) > 0)
    .mapValues(order -> {
        order.setStatus("VALIDATED");
        return order;
    });

// Branch: route to different topics by value
Map<String, KStream<String, OrderEvent>> branches = validOrders.split()
    .branch((key, order) -> "EXPRESS".equals(order.getShipping()),
            Branched.withConsumer(s -> s.to("express-orders")))
    .defaultBranch(Branched.withConsumer(s -> s.to("standard-orders")));

// Peek for side effects (logging, metrics) without mutating
validOrders.peek((key, order) ->
    log.info("Processing order {} amount {}", key, order.getAmount()));

KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();
3

Stream Processing Topology

A topology is a DAG of source, processor, and sink nodes; it is compiled from the high-level DSL (map, filter, join, aggregate) or the low-level Processor API.

  • ✓A topology is a DAG: source nodes (from topics) → processor nodes (transform/filter/aggregate) → sink nodes (to topics)
  • ✓KStream = unbounded sequence of records; KTable = last value per key (changelog semantics)
  • ✓groupBy() + count()/aggregate() produces a KTable materialised in a named RocksDB state store
  • ✓KStream-KTable join enriches each stream record with the latest value for the matching key in the table
  • ✓Kafka Streams runs as a library in your application — no separate cluster or worker nodes needed
  • ✓topology.describe() prints the full DAG structure — useful for debugging and understanding the processing graph
Java — Streams DSL: filter, map, split/branch, sink
@Configuration
public class OrderStreamConfig {

    @Bean
    public KStream<String, OrderEvent> orderStream(StreamsBuilder builder) {
        // Source: read from "order-events" topic
        KStream<String, OrderEvent> stream = builder.stream(
            "order-events",
            Consumed.with(Serdes.String(), orderEventSerde())
        );

        // Filter: only PLACED orders
        KStream<String, OrderEvent> placed = stream
            .filter((key, event) -> event.getType() == OrderEventType.PLACED);

        // Transform: enrich with metadata
        KStream<String, EnrichedOrder> enriched = placed
            .mapValues(event -> new EnrichedOrder(
                event.getOrderId(),
                event.getCustomerId(),
                Instant.now()
            ));

        // Branch: split by order value
        Map<String, KStream<String, EnrichedOrder>> branches = enriched
            .split(Named.as("branch-"))
            .branch((k, v) -> v.getTotal().compareTo(new BigDecimal("1000")) > 0,
                    Branched.as("high-value"))
            .defaultBranch(Branched.as("standard"));

        // Sink: write to separate topics
        branches.get("branch-high-value").to("high-value-orders");
        branches.get("branch-standard").to("standard-orders");

        return stream;
    }
}
4

Windowing in Kafka Streams

Tumbling windows (fixed non-overlapping), hopping windows (overlapping), session windows (activity-based gaps), and sliding windows each model different time-series aggregation needs.

  • ✓Tumbling: non-overlapping fixed buckets. Hopping: overlapping windows sliding forward. Session: inactivity-gap based.
  • ✓Grace period allows late-arriving events to update already-closed windows — critical for out-of-order data.
  • ✓Kafka Streams uses event time (record timestamp) by default — tolerates out-of-order delivery correctly.
  • ✓After the grace period expires, late events are silently dropped — monitor records-late-arrival metric.
  • ✓Session windows merge when two sessions are closer than the inactivity gap — state store merges must be idempotent.
  • ✓Use TopologyTestDriver with explicit timestamps to unit-test windowing logic without real Kafka.
Java — tumbling and hopping window aggregations
StreamsBuilder builder = new StreamsBuilder();
KStream<String, OrderEvent> orders = builder.stream("order-events");

// TUMBLING window — count orders per product per 1-hour bucket
// Windows: [00:00–01:00], [01:00–02:00], [02:00–03:00] ...
KTable<Windowed<String>, Long> hourlyCounts = orders
    .groupBy((key, order) -> order.getProductId())
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
    // Grace period: accept late events for 5 minutes after window closes
    // .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofHours(1), Duration.ofMinutes(5)))
    .count(Materialized.as("hourly-product-counts"));

// Read windowed results
hourlyCounts.toStream()
    .map((windowedKey, count) -> {
        String productId = windowedKey.key();
        long windowStart = windowedKey.window().start();
        long windowEnd   = windowedKey.window().end();
        return KeyValue.pair(productId,
            String.format("%s: %d orders in %d–%d", productId, count,
                          windowStart, windowEnd));
    })
    .to("hourly-product-counts-output");

// HOPPING window — 5-minute sum, updated every 1 minute (overlapping)
// Windows: [0:00–0:05], [0:01–0:06], [0:02–0:07] ...
orders.groupBy((k, v) -> v.getProductId())
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))
                            .advanceBy(Duration.ofMinutes(1)))
    .count();
5

State Stores & RocksDB

Stateful operations (aggregations, joins) persist state in RocksDB-backed changelog topics; state is restored automatically on restart, enabling fault-tolerant stateful processing.

  • ✓State stores back stateful operations (aggregations, joins) with local RocksDB storage and a Kafka changelog topic
  • ✓Changelog topics are compacted Kafka topics auto-created with name <app-id>-<store-name>-changelog
  • ✓On restart, state is restored by replaying the changelog — this is what makes Kafka Streams fault-tolerant without external DB
  • ✓In-memory stores (Stores.inMemoryKeyValueStore) are faster but lose state on restart and must replay full changelog
  • ✓Standby replicas (num.standby.replicas) pre-build state on secondary instances, reducing restoration time on failover
  • ✓Interactive Queries API (streams.store(…)) lets you query local state stores directly as a read model
Java — persistent state store backed by RocksDB with changelog topic
// Word count with persistent state store (default)
StreamsBuilder builder = new StreamsBuilder();

KTable<String, Long> wordCounts = builder
    .stream("sentences-input", Consumed.with(Serdes.String(), Serdes.String()))
    .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
    .groupBy((key, word) -> word)
    .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("word-count-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long()));
        // ↑ "word-count-store" is backed by RocksDB
        // changelog topic: myapp-word-count-store-changelog

wordCounts.toStream().to("word-counts-output");

// application.properties for standby replicas
// spring.kafka.streams.properties.num.standby.replicas=1
// Standby instance keeps an up-to-date copy for fast failover
6

Interactive Queries

Interactive Queries expose state store contents over a REST API, turning the Kafka Streams instance into a queryable read model without an external DB.

  • ✓State stores must be materialized with a name (Materialized.as("name")) to be queryable — anonymous stores are not accessible
  • ✓streams.store(StoreQueryParameters.fromNameAndType(...)) returns a read-only view of the local state store
  • ✓In multi-instance deployments, each instance owns only its assigned partitions — use queryMetadataForKey to find the owner
  • ✓Configure application.server=host:port so instances can discover each other for distributed query proxying
  • ✓WindowStore supports time-ranged fetch queries — retrieve aggregations for a key over a specific time range
  • ✓Interactive Queries eliminate the need to sink aggregated state to an external database for simple read models
Java — materialized state store and REST query endpoint
// 1. Materialize the state store with a name during topology build
StreamsBuilder builder = new StreamsBuilder();

KTable<String, Long> wordCounts = builder
    .stream("text-input", Consumed.with(Serdes.String(), Serdes.String()))
    .flatMapValues(v -> Arrays.asList(v.split("\\s+")))
    .groupBy((k, word) -> word)
    .count(Materialized.as("word-count-store"));  // named → queryable

// 2. Query the store from a REST endpoint
@RestController
public class WordCountController {

    private final KafkaStreams streams;

    @GetMapping("/counts/{word}")
    public Long getCount(@PathVariable String word) {
        ReadOnlyKeyValueStore<String, Long> store = streams.store(
            StoreQueryParameters.fromNameAndType(
                "word-count-store",
                QueryableStoreTypes.keyValueStore())
        );
        Long count = store.get(word);
        return count != null ? count : 0L;
    }

    @GetMapping("/counts")
    public Map<String, Long> getAllCounts() {
        ReadOnlyKeyValueStore<String, Long> store = streams.store(
            StoreQueryParameters.fromNameAndType(
                "word-count-store", QueryableStoreTypes.keyValueStore()));
        Map<String, Long> result = new HashMap<>();
        try (KeyValueIterator<String, Long> it = store.all()) {
            it.forEachRemaining(kv -> result.put(kv.key, kv.value));
        }
        return result;
    }
}
7

Kafka Streams State Stores

State stores are local RocksDB databases embedded in a Streams application enabling stateful operations like aggregations, joins, and windowed computations.

  • ✓State stores are local RocksDB instances backed by Kafka changelog topics
  • ✓Fault tolerance: state restored from changelog on task reassignment
  • ✓Interactive queries allow external APIs to read state in real time
  • ✓In-memory stores are faster but lose state on restart
  • ✓num.standby.replicas pre-warms state on standby tasks for fast recovery
Kafka Streams — aggregation + interactive query
StreamsBuilder builder = new StreamsBuilder();
KStream<String, OrderEvent> orders = builder.stream("order-events");

KTable<String, CustomerStats> stats = orders
    .groupByKey()
    .aggregate(
        CustomerStats::new,
        (customerId, event, agg) -> { agg.add(event); return agg; },
        Materialized.<String, CustomerStats, KeyValueStore<Bytes, byte[]>>
            as("customer-stats-store")
            .withValueSerde(new CustomerStatsSerde())
    );

// Read state store from REST endpoint
@GetMapping("/stats/{customerId}")
public CustomerStats getStats(@PathVariable String customerId) {
    ReadOnlyKeyValueStore<String, CustomerStats> store =
        streams.store(StoreQueryParameters.fromNameAndType(
            "customer-stats-store", QueryableStoreTypes.keyValueStore()));
    return store.get(customerId);
}
8

Kafka Streams Windowing

Windowing groups events into time-bounded buckets for aggregation. Kafka Streams supports tumbling, hopping, sliding, and session windows for real-time analytics.

  • ✓Tumbling: fixed, non-overlapping — "count per minute"
  • ✓Hopping: fixed, overlapping — "5-min rolling window, updated every minute"
  • ✓Session: closes after inactivity gap — perfect for user sessions
  • ✓Grace period allows late records; after grace, late records are dropped
  • ✓Sliding windows update continuously with every event arrival
Kafka Streams — window types
KStream<String, PageView> views = builder.stream("page-views");

// Tumbling — count per 1-minute bucket
KTable<Windowed<String>, Long> minuteCounts = views
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
    .count(Materialized.as("views-per-minute"));

// Hopping — 5-min window advancing every 1 min
KTable<Windowed<String>, Long> hoppingCounts = views
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30))
        .advanceBy(Duration.ofMinutes(1)))
    .count();

// Session — user session with 30-min inactivity gap
KTable<Windowed<String>, Long> sessionCounts = views
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapAndGrace(
        Duration.ofMinutes(30), Duration.ofMinutes(5)))
    .count();
Learn this free with Aria, your AI tutor → AiCanCode.org/learn/kafka