streams

Active contributors: Matthias J. Sax, Bill Bejeck, Alieh Saeedi, Lucas Brutschy, Mickael Maison

Purpose

Kafka Streams is a JVM library for building stream-processing applications on top of Kafka. A Streams app is a regular Java program that compiles a Topology of source nodes, processor nodes, and sink nodes; the runtime turns that topology into Kafka consumers, producers, and (for stateful processors) state stores backed by changelog topics. There is no Streams "server" — the library runs inside your application process and uses brokers for transport and durability.

Directory layout

streams/
├── src/main/java/org/apache/kafka/streams/
│   ├── KafkaStreams.java               ← top-level lifecycle (start, close, state listener)
│   ├── StreamsBuilder.java             ← high-level DSL entry point
│   ├── Topology.java                   ← lower-level processor API entry point
│   ├── StreamsConfig.java              ← configuration (~3K LOC)
│   ├── kstream/                        ← DSL types (KStream, KTable, GlobalKTable, ...)
│   ├── processor/                      ← Processor API (StateStore, ProcessorContext, ...)
│   │   ├── api/                        ← Processor / ProcessorContext public API (post-3.0 redesign)
│   │   ├── assignment/                 ← partition assignors for the rebalance protocol
│   │   └── internals/                  ← StreamThread, StreamTask, TaskManager, ...
│   ├── state/                          ← public Stores DSL + State store implementations (RocksDB, in-memory)
│   ├── query/                          ← Interactive Queries v2 API
│   ├── errors/                         ← public exception types
│   └── internals/                      ← non-public glue
├── streams-scala/                      ← idiomatic Scala wrappers
├── test-utils/                         ← TopologyTestDriver and friends
├── integration-tests/                  ← embedded-broker integration tests
├── examples/                           ← runnable WordCount, PageView, etc.
├── quickstart/                         ← Maven archetype published as a separate artifact
└── upgrade-system-tests-*/             ← one module per supported upgrade path (0.11 ... 4.x)

DSL vs Processor API

graph LR
    SB[StreamsBuilder<br/>DSL] --> TP[Topology]
    PA[Topology.addProcessor<br/>Processor API] --> TP
    TP --> KS[KafkaStreams.start]
    KS --> ST[N x StreamThread]
    ST --> SK[Per-partition StreamTask]
    SK --> RB[RocksDB / in-memory store]
    SK --> CN[KafkaConsumer]
    SK --> PR[KafkaProducer]
    RB -.->|changelog produce| CT[(changelog topic)]

• DSL (kstream/ package) provides high-level operators — map, filter, groupByKey, count, aggregate, windowedBy, join, selectKey, peek, branch, repartition. Compiles down to a Topology.
• Processor API (processor/api/) gives raw Processor<KIn,VIn,KOut,VOut> access — implement process(), register state stores manually, schedule Punctuators.
• Streams DSL — Scala (streams-scala/) is a thin wrapper that uses Scala collection idioms and implicit Serdes.

Key abstractions

Type	File	Purpose
KafkaStreams	streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java	App-level lifecycle: start, close, state listener, metrics, IQ.
StreamsBuilder / Topology	streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java, Topology.java	DSL / Processor API entry points.
KStream, KTable, GlobalKTable	streams/src/main/java/org/apache/kafka/streams/kstream/	Continuous record stream / changelog table abstractions.
Processor / ProcessorContext / Punctuator	streams/src/main/java/org/apache/kafka/streams/processor/api/	Processor API (post-3.0 redesign).
StreamThread	streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java	One thread runs N tasks; owns a consumer + producer pair.
StreamTask / StandbyTask	streams/.../processor/internals/StreamTask.java, StandbyTask.java	Active and standby per-partition unit of work.
TaskManager	streams/.../processor/internals/TaskManager.java	Owns task lifecycle, restoration, suspension, recycling.
StreamsPartitionAssignor (classic) and KIP-848-aware variants	streams/.../processor/internals/StreamsPartitionAssignor.java, streams/.../processor/assignment/	Computes active + standby task assignments across instances.
StateStore / KeyValueStore / WindowStore / SessionStore	streams/src/main/java/org/apache/kafka/streams/state/	State store API; default impls are RocksDB-backed.
KeyValueIterator, Stores	streams/src/main/java/org/apache/kafka/streams/state/Stores.java	Builders for stores.
TopologyTestDriver	streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java	Deterministic single-thread test harness.
StreamsConfig	streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java	Configuration; embeds producer/consumer prefixes (producer.*, consumer.*).

Threads, tasks, and partitions

A Streams app instance is configured with num.stream.threads. Each StreamThread:

1. Owns a KafkaConsumer and a KafkaProducer.
2. Is assigned a set of StreamTasks (one per input partition group) plus optional StandbyTasks.
3. In its loop, polls the consumer, advances each task by feeding records into its sub-topology, periodically commits, and runs Punctuators.

Stateful operators (groupByKey().count(), joins, windowed aggregates) require state stores. Each store's writes are produced to a Kafka changelog topic so that a task moving to a different instance can rebuild the store from that topic. Standbys keep changelogs continuously hot-warm to reduce rebalance time.

Rebalance and assignment

Streams uses a custom PartitionAssignor that is aware of:

• which input topics each subtopology consumes,
• per-instance hardware tags (group.instance.id, client.tag.*),
• previous assignments (sticky),
• standby task placement.

The classic protocol delegates assignment to a designated leader client. Under KIP-848 + the new Streams group protocol (KIP-1071, in-flight on trunk), assignment moves into the broker via a Streams-specific PartitionAssignor registered with the group coordinator.

Exactly-once

Setting processing.guarantee=exactly_once_v2 makes each StreamTask use a transactional KafkaProducer (transactional.id derived per task). The task commits input offsets and output records in one transaction per commit interval. Internally, the task uses the producer's sendOffsetsToTransaction to atomically move offsets and produce changelog updates. See features/exactly-once.md.

Interactive queries

KafkaStreams.store(...) (legacy) and KafkaStreams.query(...) (the v2 API in streams/.../query/) let an application read state stores directly. StreamsMetadata describes which instance currently hosts a given store / partition so that callers can route queries.

State store backends

The default state store is RocksDB (streams/.../state/internals/RocksDBStore.java); in-memory variants are also provided (InMemoryKeyValueStore, etc.). Stores are wrapped in layers:

MeteredStore   ← metrics
  CachingStore  ← write-back cache (optional, controlled by cache.max.bytes.buffering)
    ChangeLoggingStore ← writes to the changelog producer
      RocksDBStore / InMemoryStore  ← actual storage

Each layer is a WrappedStateStore decorator on the next.

TopologyTestDriver

Use this for unit tests instead of a full integration test:

TopologyTestDriver driver = new TopologyTestDriver(topology, props);
TestInputTopic<String, String> in = driver.createInputTopic("input", new StringSerializer(), new StringSerializer());
TestOutputTopic<String, Long> out = driver.createOutputTopic("output", new StringDeserializer(), new LongDeserializer());
in.pipeInput("key", "v1");
assertThat(out.readKeyValue()).isEqualTo(new KeyValue<>("key", 1L));

It runs the topology in-process with a controllable mock time and simulated Kafka. See streams/test-utils/.

Configuration

StreamsConfig declares Streams-specific options and embeds producer/consumer/admin properties via producer.*, consumer.*, admin.*, main.consumer.*, restore.consumer.*, global.consumer.* prefixes. Key options:

• application.id — group ID for the embedded consumer; namespace for internal topics.
• bootstrap.servers.
• processing.guarantee — at_least_once (default) or exactly_once_v2.
• num.stream.threads, num.standby.replicas, num.warmup.replicas (KIP-1102, recently added).
• default.key.serde, default.value.serde.
• cache.max.bytes.buffering, commit.interval.ms.
• group.protocol — classic or streams (KIP-1071).

Entry points for modification

• DSL operators: streams/.../kstream/internals/. Each public DSL method maps to a KStreamImpl / KTableImpl method that adds a processor node to the topology.
• Processor API: streams/.../processor/api/ for the public surface, streams/.../processor/internals/ for the runtime.
• Rebalance / assignment: StreamsPartitionAssignor.java and the streams/.../processor/assignment/ package.
• New state store type: implement StateStore, register a Stores.<type>StoreBuilder, optionally provide a custom changelog supplier.
• Streams group protocol (KIP-1071): coordinator assignor lives in group-coordinator/.../coordinator/group/streams/.

Related pages

• Modules: clients — Streams runs on top of the producer / consumer / admin clients.
• Modules: coordinators — group coordinator runs Streams group rebalances.
• Features: Exactly-once semantics.