Kafka Streams

A consumer that reads a topic, transforms each record and writes to another topic is a stream processor, and writing one by hand means solving the same problems every time: state that survives a restart, windows over time, joins between topics, and exactly-once between input and output. Kafka Streams is a library, not a cluster, that solves them inside an ordinary JVM application, using Kafka itself for its state and its coordination. This lesson is the two abstractions, where state lives, windows, joins, and what exactly-once means here.

#KStream and KTable

A KStream is a topic read as an unbounded sequence of events: every record is a fact that happened, and two records with the same key are two facts. A KTable is a topic read as a changelog: every record is the latest value for its key, and a new record with the same key replaces the old one. The same bytes can be either; the choice is about meaning. Orders placed are a stream; the current status of each order is a table; a user's profile is a table; clicks are a stream.

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Order> orders = builder.stream("orders");
KTable<String, Customer> customers = builder.table("customers");     // compacted topic, keyed by customer id
 
KStream<String, EnrichedOrder> enriched = orders
    .selectKey((orderId, order) -> order.customerId())               // re-key to join: triggers a repartition
    .join(customers, (order, customer) -> EnrichedOrder.of(order, customer.tier()))
    .filter((customerId, e) -> e.tier() != Tier.BLOCKED);
 
enriched.to("orders-enriched");

A topology is the graph of these operations, built once and run by KafkaStreams. Stateless operations (map, filter, branch, flatMap) are per record and need nothing. Everything interesting is stateful.

A GlobalKTable is a table replicated in full into every instance, rather than partitioned across them; it is for small reference data (currencies, tiers, feature flags) that any record might need, and it removes the co-partitioning requirement below at the cost of every instance holding the whole table.

#State stores

count(), aggregate(), reduce() and every join need to remember things between records. Streams keeps that state in a state store: a RocksDB instance on the local disk of the instance that owns the partition, with a changelog topic in Kafka receiving every change. When an instance dies, the partition and its state move to another instance, which rebuilds the store by replaying the changelog. That replay is the recovery time, and for a large store it is minutes; num.standby.replicas=1 keeps a warm copy on a second instance so failover is seconds instead.

KTable<String, Long> ordersPerCustomer = orders
    .groupBy((orderId, order) -> order.customerId())
    .count(Materialized.as("orders-per-customer"));        // a named, queryable store

A named store can be read by the application through interactive queries, which makes a Streams app a serving layer for its own aggregates without a database in between; each instance answers for its own partitions and can route to the instance that holds another key.

State stores are why a Streams instance is not stateless and cannot be treated like a web pod: it has a disk, it takes time to be ready after a restart, and scaling it in or out moves partitions and their state. Static membership (group.instance.id, from the Consumer groups lesson) and standby replicas are how a deployment makes that tolerable.

#Windows

An aggregate over a stream is infinite unless bounded by time. Windows bound it:

KTable<Windowed<String>, Long> failures = logins
    .filter((user, attempt) -> !attempt.ok())
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1)))
    .count();
failures.toStream().filter((w, n) -> n >= 5).to("suspicious-logins");

Time is the record's event time (the timestamp the producer set) not the time it arrived, so a record that was delayed lands in the window it belongs to. The grace period is how long after a window's end a late record is still accepted; after it, the window is closed and late records are dropped. A grace of zero is the fastest and drops everything late; a long grace holds windows open in the store longer. Choose it from how late your producers actually run, measured, not guessed.

Window results are emitted as they update, so a downstream consumer sees the count go 1, 2, 3, 4, 5 for one window. suppress(untilWindowCloses(...)) emits only the final result, once, at the cost of the grace period's latency, and is what an alerting pipeline wants.

#Joins

Co-partitioning is the rule that makes joins possible without a network: both inputs must have the same number of partitions and be keyed the same way, so that key K is in partition 3 on both sides and one instance holds both. selectKey on a stream that then joins triggers a repartition topic, an internal topic Streams creates to re-key and redistribute the records; it is automatic, correct, and doubles the traffic for that stream, which is why a topology's repartitions are worth counting.

#Exactly-once

processing.guarantee=exactly_once_v2 turns the whole read–process–write cycle into the transaction from the Offsets lesson: input offsets, changelog writes for state stores, and output records commit together, and a crash mid-cycle replays cleanly with no duplicated output and no double-counted state. This is the strongest guarantee Kafka offers and Streams is where it is practical, because everything the job touches is a Kafka topic. The moment a topology calls out to something else (a database in a map, an HTTP client in a foreach), that call is outside the transaction and the idempotency rules return. The cost is latency: commit.interval.ms drops to 100 ms by default under exactly-once, and every commit is a transaction.

#When Streams, and when not

Streams is right for transforming, enriching, aggregating and joining Kafka topics into other Kafka topics, inside a service you already run, with no cluster to operate. Plain consumers are right when the work is "read a record, call a service, write to a database", which Streams makes no easier. Flink or Spark are right when the processing needs a cluster of its own: sources that are not Kafka, very large state, or SQL over streams for people who are not writing Java. Most backend teams need the first two and should think hard before the third.

#Under the hood: tasks, the changelog, and the flush-then-commit cycle

KafkaStreams.start() turns the topology into tasks: one task per input partition (for co-partitioned inputs, one task per partition number), each owning that partition's slice of every state store and every downstream sub-topology up to a repartition topic, where a new sub-topology and its own tasks begin. Tasks are assigned across num.stream.threads threads in each instance and across instances through an ordinary consumer group (application.id is the group id), with a Streams-specific assignor that prefers to place a task where its state already lives. Each thread runs one consumer and one producer and loops: poll a batch, hand each record through the task's processor chain (map, filter, aggregation processors that read and write the task's stores), and periodically commit: flush the state stores' RocksDB memtables, flush the producer so every changelog and output record is acknowledged, then commit the input offsets. That order is the guarantee: state is durable in the changelog before the offsets that produced it are committed, so a crash replays at most the uncommitted window and the changelog replay reproduces the state.

A state store is three things layered. A RocksDB instance under state.dir/<app-id>/<task-id>/, an in-memory record cache in front of it (statestore.cache.max.bytes, formerly cache.max.bytes.buffering) that absorbs repeated updates to the same key and emits downstream only on flush or eviction, which is why a count() produces fewer output records than input records, and the changelog topic, compacted, written by the task's producer on every store update. Restoration on failover is a consumer reading that changelog from its beginning (or from the task's local checkpoint file, if the disk survived) into RocksDB; standby tasks (num.standby.replicas) keep another instance's copy warm by consuming the changelog continuously, so promotion is a checkpoint away.

Under exactly_once_v2 each stream thread's producer is transactional (one transactional.id per thread since v2, rather than per task as in v1, which is what made it scale), and the commit cycle becomes: flush stores, then sendOffsetsToTransaction and commitTransaction, so changelog writes, output records and input offsets are one atomic unit. commit.interval.ms drops to 100 ms because every uncommitted window is latency for read_committed downstream consumers. A crash mid-transaction is aborted by the fencing on restart, the changelog beyond the last committed transaction is ignored during restore (Streams tracks the committed changelog offset in the checkpoint), and RocksDB is rolled back by wiping and restoring, since RocksDB itself is not transactional with the changelog.

Time is a per-task stream time, the maximum record timestamp seen so far, advanced only by records. Windows close relative to stream time, not the wall clock, so a partition that stops receiving records stops closing windows, and suppress holds results until stream time passes the window end plus grace. Punctuators can be scheduled on wall-clock time for the cases that need it.

#Walkthrough: the count that came back wrong after a scale-out

A fraud-scoring topology counted failed logins per account in one-hour tumbling windows and alerted at five. After scaling from three instances to six, alerts stopped for twenty minutes, then a handful fired for accounts that had two failures, not five.

logins.filter((k, v) -> !v.ok())
      .groupByKey()
      .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofHours(1), Duration.ofMinutes(5)))
      .count(Materialized.as("failures"));        // no standby, no persistent volume, eager rebalance

1. Scaling out moved half the tasks. Under the eager protocol every instance revoked every task, and the three new instances had no local state: each restored its tasks' failures-changelog partitions from the beginning, 40 GB across them, at the changelog's read rate. Twenty minutes with no processing on the moved tasks and, since the old instances had also revoked, on the unmoved ones too.
2. The three original pods had no persistent volume, so their local RocksDB and checkpoint files were gone on the restart that accompanied the scale-out; their retained tasks restored from scratch as well.
3. The wrong alerts: the record cache had absorbed several increments to the same key without emitting; the flush that would have written them to the changelog was in the uncommitted window when the instances were killed. On restore, the changelog lacked those increments, and stream time on the restored task had advanced, so the counts under the current window were lower than they had been. The alert at five then fired for keys whose restored count crossed five later, while accounts that had truly reached five earlier did not re-alert because stream time had moved past their window plus grace, and suppress had never emitted.
4. Fixes: num.standby.replicas=1 so every task has a warm copy and scale-out promotes rather than restores; a persistent volume per pod with group.instance.id so a restart reuses its local state via the checkpoint; the cooperative assignor (the Streams default since 2.6, but the app had pinned an old assignor config); and exactly_once_v2, so cache flushes, changelog writes and offsets commit together and a kill cannot leave the changelog behind the state.
5. The verification was a chaos test in staging: kill an instance during load and assert that windowed counts on the survivor match a batch recount from the input topic.

Streams state is only as durable as the changelog is current, and only as quick to move as the standby is warm. A Streams app deployed like a stateless service is a stateful service that has agreed to lose its state at every deploy.

#Try it yourself

#Misconceptions

• "A Streams app is a stateless consumer with extra operators." It is a set of tasks each owning local state; without volumes, standbys and static membership, every deploy is a restore.
• "State is in Kafka, so it is safe." State is in RocksDB and in the changelog as of the last flush and commit. Under at-least-once the two can diverge across a crash; exactly_once_v2 makes them commit together.
• "Windows close on the clock." They close on stream time, advanced by records. A quiet partition never closes its windows.
• "count() emits one record per input." The record cache coalesces updates per key and emits on flush; downstream sees fewer, later updates unless the cache is disabled.
• "Exactly-once means my side effects happen once." It means Kafka writes and offsets are atomic. A database or HTTP call inside the topology is at-least-once.

#Going deeper

• Kafka Streams documentation: "Architecture" (tasks, threads, state, fault tolerance) and "Developer Guide → Processing Guarantees" and "Memory Management" (the record cache).
• KIP-129 (Streams exactly-once), KIP-447 (exactly_once_v2) and KIP-441 (smooth scaling, warm-up replicas).
• Matthias J. Sax, "Streams and Tables in Apache Kafka" and Michael Noll's "Streams and Tables: A Primer" series.
• Confluent, "Kafka Streams' Take on Watermarks and Triggers", on stream time, grace and suppress.
• Mitch Seymour, Mastering Kafka Streams and ksqlDB.