Kafka Streams

A KStream is a record stream where every record is an independent event (insert/append). A KTable is a changelog stream where each record is an update keyed by its key — only the latest value per key matters.

A KTable is partitioned: each task holds only its partition's keys. A GlobalKTable is fully replicated: every instance holds ALL partitions of the source topic. So GlobalKTable joins need no co-partitioning and can join on a non-key field.

StreamsBuilder is the DSL entry point. builder.stream(topic) returns a KStream (event stream), builder.table(topic) returns a KTable (latest-value-per-key, materialized), and builder.globalTable(topic) returns a GlobalKTable (fully replicated). builder.build() produces the Topology.

A topology is split into sub-topologies at points where data must be re-shuffled by key — repartition topics. Operations that change the key (selectKey, map, groupBy) force a repartition; the topology breaks there into independently-scheduled sub-topologies.

Call topology.describe() to print the processor graph (sub-topologies, sources, sinks, stores). Naming operators (via Named/Materialized/Repartitioned/Joined) gives stable node and internal-topic names, so refactoring the DSL doesn't change changelog/repartition topic names and break state restore.

A stateless operation processes each record independently, keeping no memory of past records. Examples: map, mapValues, filter, filterNot, flatMap, flatMapValues, selectKey, branch/split, merge, peek, foreach.

map can change both key and value; mapValues changes only the value and keeps the key. Prefer mapValues because keeping the key avoids a repartition topic when a stateful operation follows.

Key-changing stateless ops (selectKey, map, flatMap) set a repartition-required flag. When a stateful op follows, Streams injects an internal repartition topic (named app-id + node + '-repartition') to re-shuffle by the new key. Use mapValues to avoid it, or repartition() to control it.

Use KStream.split() then chained .branch(predicate, Branched.as("name")) and optionally .defaultBranch(), which returns a Map<String, KStream> of routed sub-streams. merge(otherStream) does the opposite: it interleaves two same-typed streams into one. Both are stateless.

flatMap/flatMapValues turn one input record into zero-or-many output records (flatMapValues keeps the key). peek runs a side effect per record and passes records through unchanged. foreach is terminal: it runs a side effect and ends the branch with no downstream. Side effects aren't transactional, so they can repeat on retries.

A state store is a local key-value store on each Kafka Streams instance that holds data needed for stateful operations like aggregations, joins, and windowing — so the app can remember results across messages instead of treating each event in isolation.

Every update to a state store is also written to a special Kafka topic called a changelog. The changelog is the durable backup: if an instance dies, another instance replays the changelog to rebuild the store. Changelogs are log-compacted so they stay bounded.

In the DSL you configure stores with Materialized — passing it to operators like count() or aggregate() to set the store name, serdes, and persistence. In the lower-level Processor API you build a store explicitly with the Stores factory and a StoreBuilder, then register it on the topology.

Each state store has an in-memory record cache that deduplicates and batches updates. Downstream operators and the changelog only see the latest value per key when the cache is flushed — which happens when the cache fills up or at every commit (commit.interval.ms). Disabling the cache makes every update flow through immediately.

When tasks move between instances during a rebalance, the new owner must rebuild each store by replaying its changelog topic before it can process records. This restore can be slow. Standby replicas keep warm copies on other instances to avoid or shorten it, and a persistent state.dir lets an instance reuse local state instead of restoring from scratch.

Kafka Streams supports stream-stream joins (over a time window), stream-table joins (enrich a record with the latest table value), and table-table joins (combine two changelog tables). GlobalKTable joins let a stream join a fully-replicated table without matching partitioning.

Co-partitioning means both join inputs must use the same key, the same number of partitions, and the same partitioning strategy, so matching keys land on the same task. You satisfy it by re-keying and repartitioning; GlobalKTable joins avoid it entirely.

A stream-table join enriches each stream record with the table's current value for that key. Only stream records trigger output (table updates don't). Inner emits only on a match; left emits the stream record paired with null when no match. GlobalKTable joins skip co-partitioning and join on any value-derived key.

A foreign-key join lets a left KTable join a right KTable on a key extracted from the left value, not the left's own key. Kafka Streams repartitions the left table by the foreign key internally, so the two tables don't need to be co-partitioned, and updates on either side correctly propagate.

A JoinWindows defines a time difference: record A joins record B if their timestamps are within the window. The grace period allows late records before the window closes. Inner results emit immediately on a match; left/outer null-paired results emit only after the window plus grace expires.

Tumbling (fixed, non-overlapping), hopping (fixed size, overlapping by advance interval), sliding (window around record pairs within a time difference), and session (activity-gap-based, dynamic size). The first three are time-based; sessions are data-driven.

The grace period is extra time after a window's end during which late (out-of-order) records are still accepted and update the window's result. Once stream time passes window-end + grace, the window is closed and later records are dropped.

JoinWindows define how close in event time two records from the two streams must be to join. With JoinWindows.ofTimeDifferenceAndGrace(d), records join if their timestamps differ by at most d (symmetric: a within [b-d, b+d]). Each stream is buffered in a window store for that span.

A session window groups records for a key that arrive within an inactivity gap of each other. The window grows with each new record and closes when no record arrives for longer than the gap. Out-of-order records can merge two adjacent sessions into one.

A windowed aggregation produces a KTable keyed by Windowed<K> (the original key plus the window's start/end). It is stored in a segmented WindowStore (RocksDB split into time segments) and serialized with a WindowedSerdes that encodes key + window timestamp. Old segments are dropped wholesale when they fall outside retention.

Event time = when the event actually happened (set by the producer). Ingestion time = when the broker appended the record. Processing time = when Kafka Streams processes it. They differ because of network and processing delays.

TimestampExtractor tells Streams which timestamp to use per record. FailOnInvalidTimestamp (default) reads the record timestamp and throws on a negative one; LogAndSkipOnInvalidTimestamp logs and skips bad ones; WallclockTimestampExtractor ignores the record and uses System.currentTimeMillis().

Stream time is the app's notion of the current event-time clock per task: the maximum record timestamp seen so far. It only moves forward — a later record with an older timestamp does not pull it back. It drives window closing and lateness decisions.

Out-of-order records (timestamp below stream time but within the grace period) are still added to their window and update the result. Once stream time passes window end plus the grace period, the window closes and later-arriving records for it are dropped as 'late'.

When a task reads several partitions, max.task.idle.ms tells Streams how long to pause/wait for an empty-but-active partition to deliver data before processing ahead with other partitions. It trades join/merge time-ordering correctness against latency.

groupByKey() groups by the existing record key with no repartition. groupBy() picks a NEW key, so Streams must repartition (re-shuffle records to topic partitions) so all records with the same new key land on the same task.

count() returns how many records per key. reduce() combines two values of the SAME type into one (e.g. sum of longs). aggregate() is the general form: an Initializer creates a starting value of ANY type, and an Aggregator folds each record into it — use it when the result type differs from the input.

By default a windowed aggregation emits an update on every record, so downstream sees many intermediate counts per window. suppress(Suppressed.untilWindowCloses(...)) buffers updates and emits only the FINAL result for each window after the window plus grace period has closed. It needs a buffer config (e.g. unbounded or a bytes/records limit) and only works on windowed KTables.

You call windowedBy(...) on a KGroupedStream before count/reduce/aggregate, so records are bucketed by time. The result is a KTable keyed by Windowed<K> (key + window bounds). Window types include tumbling, hopping, sliding, and session windows. Late records that arrive within the grace period still update their window; beyond grace they are dropped.

A KTable is a changelog: each key maps to its latest aggregate. After every input record, the aggregate for that key changes, so the KTable emits a new (key, newAggregate) update. Downstream therefore sees the running aggregate update again and again, not just the final value.

The Processor API (PAPI) is the low-level Kafka Streams API. You write a Processor that handles each record one at a time, can attach state stores, and forward results downstream. The DSL is a higher-level, declarative layer (map, filter, join) built on top of it.

context.forward(record) sends a record from your processor to its downstream child nodes. You can call it zero, one, or many times per input record. Passing forward(record, "childName") routes it to only one named child instead of all of them.

schedule() registers a punctuator — a callback that fires periodically. STREAM_TIME advances by the timestamps of records flowing through, so it only fires when data arrives and progresses. WALL_CLOCK_TIME advances by the system clock, so it fires on a real-time interval even if no records arrive.

Register the store with a StoreBuilder (addStateStore in PAPI, or builder.addStateStore in the DSL), connect it to the processor by name, then look it up in init() via context.getStateStore("name"). In the DSL, pass the store names as the second argument to process(supplier, "storeName").

A ProcessorSupplier.get() returns a new Processor for each task/thread, so instances aren't shared across threads. init() runs once per task to grab the context and stores, process() runs per record, and close() runs once when the task shuts down to release resources.

Interactive Queries let your app read directly from the state stores Kafka Streams already maintains, instead of querying an external database. You call KafkaStreams.store(...) to get a read-only store and look up a key with get().

A local store only holds the partitions assigned to its instance. To find the owner of a key, call streams.queryMetadataForKey(storeName, key, serializer). It returns the host:port (from application.server) of the instance that owns that key's partition, so you can route the query there.

Run an HTTP (or gRPC) server on each instance with a query endpoint. On a request, use queryMetadataForKey to find the owning host. If it's local, read the local store and return the value; if remote, forward the request to that host's endpoint and relay the result.

Setting num.standby.replicas > 0 makes other instances keep warm copies of a store's state. If the active instance fails, queryMetadataForKey still lists standby hosts, so you can serve reads from a standby for higher availability — but a standby may be slightly behind, so its data can be stale.

IQv2 is a newer, type-safe query API (StateQueryRequest + Query types like KeyQuery/RangeQuery) returning a StateQueryResult, designed to be extensible and partition-aware. Windowed stores use a different queryable type (windowStore/sessionStore) and you query by key plus a time range, not just a key.

It guarantees each input record affects the application's state and output exactly once, even if the app crashes and restarts — no duplicates and no lost updates. You enable it by setting the config processing.guarantee to exactly_once_v2.

Streams uses a single Kafka transaction per commit. The output writes, changelog (state) writes, and the consumed input offsets are all added to that transaction and committed together via sendOffsetsToTransaction + commitTransaction, so they advance as one atomic unit.

The downstream consumer is almost certainly reading with isolation.level=read_uncommitted (the default), so it sees records from aborted transactions and uncommitted writes. Fix: set isolation.level=read_committed on the downstream consumer so it only reads committed output.

Under EOS, commit.interval.ms defaults to 100 ms (vs 30000 ms at-least-once) because each commit ends a transaction, and downstream read_committed consumers only see output after a commit. Smaller interval = lower end-to-end latency but more commit overhead; larger interval = higher throughput but more latency and bigger replay on failure.

The original exactly_once used one transactional producer per task (input partition), which didn't scale — many producers and slow rebalances. exactly_once_v2 (KIP-447) uses one producer per stream thread and fences zombies via consumer-group metadata, so it scales far better. v2 needs brokers >= 2.5.

A task is the smallest unit of parallelism in Kafka Streams. The number of tasks equals the number of partitions of the busiest (most-partitioned) input topic in the topology. Each task processes a fixed set of partitions.

num.stream.threads sets how many StreamThreads run inside one application instance. Tasks are distributed across all threads of all instances. You scale up (more threads per instance) or out (more instances), but never beyond the total number of tasks.

num.standby.replicas keeps shadow copies of a stateful task's state store on other instances, continuously fed from the changelog topic. On failover, an up-to-date standby takes over almost immediately instead of rebuilding state from scratch, cutting recovery time.

Streams auto-creates two kinds of internal topics: repartition topics (to re-shuffle data by a new key before stateful ops) and changelog topics (a durable backup of each state store). Their partition counts mirror the input, which also bounds parallelism.

Cooperative rebalancing (the default since 2.4) lets instances keep tasks they already own during a rebalance instead of giving everything up. Only the tasks that must move are revoked, avoiding a global stop-the-world pause and unnecessary state restoration.