Stream Processing

Stream processing lets you react to events as they happen instead of waiting for the next batch job to run. For platform engineers, SREs, and backend developers running Apache Kafka in production, it is an operational discipline: stateful processes that hold local state on disk, restore that state after rebalances, and must keep consumer lag near zero under load. This pillar page covers the architecture, the stateful machinery, the time and correctness semantics, and the day-two operational concerns of running stream processing on Kafka.

The model is simple to state and harder to operate. Rather than treating data as static records queried after the fact, stream processing treats it as an unbounded flow of events that you transform, enrich, aggregate, and join as they arrive. Kafka provides the durable, partitioned, replicated log underneath; Kafka Streams, ksqlDB, and Apache Flink provide the processing semantics on top. The hard parts—partitioning, local state, windowing, exactly-once delivery, rebalancing, and observability—are where production teams spend their time.

The Architecture of Stream Processing on Kafka

At its core, stream processing on Kafka means consuming records from input topics, applying transformations, and producing results to output topics. Kafka’s partitioned topic model provides the parallelism: each partition is processed independently, enabling horizontal scaling without cross-instance coordination. The consumer group protocol distributes partitions across instances and rebalances them when instances join or leave.

The coupling between Kafka’s storage layer and the processing engine drives much of the operational behavior. Because Kafka durably retains records on disk, processors can reprocess history, recover from failures, and join streams that arrive at different times. When a Kafka Streams application starts, it restores its state stores from changelog topics held in Kafka, reconstructing local state without an external database. That tight coupling simplifies the architecture but makes topic retention, disk provisioning, and state-store sizing first-order operational concerns.

A key design decision is the high-level DSL versus the low-level Processor API. The DSL is declarative and covers most needs—filtering, mapping, grouping, aggregating, joining. The Processor API exposes the topology directly, giving you control over record-by-record processing, explicit state-store access, and punctuate()-based scheduling for time- or count-driven work. For a detailed comparison, see Kafka Streams DSL vs Processor API: When to Use Each.

Topology Design and Partitioning Strategy

Every Kafka Streams application defines a topology: a directed acyclic graph of processor nodes connected by streams. The topology is parallelized by Kafka’s partitioning model. Each partition of an input topic is assigned to exactly one stream task, and each task instantiates the full topology. Records with the same key always land on the same task, which is what makes local stateful operations possible without distributed coordination.

// Define a simple topology using the Kafka Streams DSL
StreamsBuilder builder = new StreamsBuilder();
KStream<String, Order> orders = builder.stream("orders-topic");
KStream<String, EnrichedOrder> enriched = orders
    .filter((key, order) -> order.getAmount() > 0)
    .mapValues(order -> enrichWithCustomerData(order))
    .selectKey((key, order) -> order.getRegionId());
enriched.to("enriched-orders-topic");

The choice of partitioning key has real operational consequences. A poorly chosen key causes skew, where one partition takes disproportionate load and becomes a hot spot that caps throughput and inflates latency. Note that selectKey and other re-keying operations mark the stream for repartitioning: Kafka Streams inserts an internal repartition topic so that downstream aggregations and joins see data co-partitioned by the new key. That extra topic is real network and storage cost, so re-key deliberately, not incidentally.

Stateful Stream Processing and Local State Stores

Stateless transformations—filtering, mapping, branching—are easy to operate because they need no storage beyond Kafka’s log. Stateful operations introduce local state stores that you must size, monitor, and recover. The common ones are aggregations, windowed computations, and stream-table joins. The operational depth here is covered in Stateful Stream Processing with Kafka Streams.

Kafka Streams uses RocksDB as its default persistent state store, maintaining an LSM-tree on local disk per stateful task. Keeping state local avoids per-record network round-trips to an external database, but it means every instance needs adequate local storage and that state must be restored after rebalances or failures. Durability comes from the changelog: each update to a persistent store is also written to a compacted Kafka topic, which is the source of truth for reconstruction.

# Kafka Streams configuration for state store management
state.dir=/data/kafka-streams/state
rocksdb.config.setter=com.example.CustomRocksDBConfig
commit.interval.ms=100
# statestore.cache.max.bytes replaces the deprecated
# cache.max.bytes.buffering (deprecated in 3.4)
statestore.cache.max.bytes=10485760

The record cache (statestore.cache.max.bytes) deduplicates and buffers updates before they hit the store and the downstream; a larger cache reduces downstream traffic and changelog writes at the cost of higher output latency, because results are emitted less frequently. commit.interval.ms interacts with this: with exactly-once enabled the default commit interval is 100 ms, and forced flushes on commit also emit cached results.

Interactive Queries and State Store Access

Kafka Streams can serve reads directly from its local state stores, avoiding a separate serving database for many lookup patterns. This is called interactive queries, and it lets an application answer low-latency point and range lookups against its latest aggregations and materialized tables. Because state is partitioned across instances, you also need a way to find which instance hosts a given key.

KafkaStreams.queryMetadataForKey(...) returns the host (from application.server) that owns a key’s partition, so a request hitting the wrong instance can be proxied to the right one over an RPC endpoint you expose yourself—Kafka Streams does not provide the RPC layer. For the full pattern, including standby replicas and routing, see Interactive Queries and State Stores in Kafka Streams.

// Query a local state store via interactive queries
ReadOnlyKeyValueStore<String, Long> countStore = streams.store(
    StoreQueryParameters.fromNameAndType(
        "count-store", QueryableStoreTypes.keyValueStore()));

Long count = countStore.get("order-123");

Stream-Table Duality and KTable Semantics

A concept that trips up engineers new to Kafka Streams is the distinction between KStream and KTable, and the stream-table duality underneath it. A KStream is a record stream: each record is an independent fact, and the stream is append-only. A KTable is a changelog stream: each record with a given key overwrites the previous value, and the table represents the latest value per key. A null value in a KTable source is a tombstone that deletes the key.

This duality means a topic can be read as either a stream or a table depending on what you need. A topic of customer-profile updates can be consumed as a KStream to track every change, or as a KTable to materialize the current profile per customer. Getting this right is essential for correct joins and aggregations. The operational consequence: a KTable state store grows with the number of distinct keys, and the source topic must use log compaction (cleanup.policy=compact) so the changelog stays bounded. For more, see Stream-Table Duality and KTable Operations.

// Demonstrating stream-table duality
KStream<String, Transaction> txnStream = builder.stream("transactions");
KTable<String, Account> accountTable = builder.table("accounts");

// KStream-KTable join enriches each transaction with the latest account
KStream<String, EnrichedTransaction> enriched = txnStream
    .join(accountTable, (txn, account) ->
        new EnrichedTransaction(txn, account.getOwnerName()));

A KStream-KTable join is non-windowed and looks up the table’s current value for the matching key. The two inputs must be co-partitioned: same number of partitions and the same partitioning, or the join silently misses matches.

Windowing and Temporal Semantics

Time is a first-class concept in stream processing, and windowing lets you aggregate over bounded intervals. Kafka Streams supports tumbling windows (fixed-size, non-overlapping), hopping windows (fixed-size, overlapping), session windows (activity-based, variable-size), and sliding windows. Sliding windows, added in 2.7.0 (KIP-450), are a fixed-size aggregation window where a new window is emitted whenever the set of records within the maximum time difference changes—more efficient than approximating with a small-advance hopping window. (A separate sliding join window governs KStream-KStream joins.)

The distinction between event time, ingestion time, and processing time matters for correctness. Event time—the timestamp the producer embedded in the record—is usually the right basis for business logic, but it forces you to handle out-of-order and late records. Kafka Streams uses a per-task stream time advanced by observed record timestamps, plus a configurable grace period, to decide when a window is closed; records arriving after the grace period are dropped and counted as late. (Grace is set in code, e.g. TimeWindows.ofSizeAndGrace(...).) ksqlDB exposes the same windowing through SQL. For examples and tuning, see Windowed Aggregations and Joins in ksqlDB.

-- ksqlDB windowed aggregation with a 5-minute tumbling window
CREATE TABLE order_counts AS
SELECT
    item_id,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue
FROM orders_stream
WINDOW TUMBLING (SIZE 5 MINUTES)
GROUP BY item_id
EMIT CHANGES;

Managing Window State and Retention

Windowed state stores grow with key cardinality and window duration. Window stores are organized into time-ordered segments; a segment is dropped once it falls outside the store’s retention, which a background thread reclaims. Retention must cover the window size plus the grace period at minimum—set it too short and you drop late records and break late-arriving queries; set it too long and you burn disk.

The window-store changelog gets cleanup.policy=compact,delete with a retention derived from the store’s retention plus windowstore.changelog.additional.retention.ms (default 24 hours), which guards against clock drift and changelog truncation while the app is offline. Note that the window size and grace period are set in code on the window definition, not as application config keys.

# Streams app config that affects windowed changelogs
windowstore.changelog.additional.retention.ms=86400000

Exactly-Once Processing Semantics

Exactly-once semantics—each input record affecting state and output exactly once, even across failures—is one of the hardest guarantees in distributed stream processing. Kafka Streams delivers it by binding state-store updates, output records, and consumer offset commits into a single Kafka transaction using the idempotent, transactional producer. The guarantee is end-to-end within Kafka: from consuming an input record, through state updates, to producing results.

Enable it with processing.guarantee=exactly_once_v2 (the constant StreamsConfig.EXACTLY_ONCE_V2). The earlier exactly_once mode is deprecated since 3.0 and removed in 4.0; exactly_once_v2 (introduced in 2.8 via KIP-447) uses a single producer per instance instead of one per task, which scales far better. It requires brokers running 2.5 or later. Under exactly-once, Kafka Streams forces isolation.level=read_committed and enable.idempotence=true and will not let you override them. The cost is real: transactional commits add latency, so the default commit.interval.ms drops to 100 ms, and the broker-side transaction coordinator becomes a component you monitor. For the operational details, see Exactly-Once Processing in Kafka Streams.

// Enabling exactly-once semantics in Kafka Streams
Properties props = new Properties();
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
    StreamsConfig.EXACTLY_ONCE_V2);
// REPLICATION_FACTOR_CONFIG defaults to -1 (use broker default) since 3.0;
// set it explicitly only if you want to override the broker default
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
props.put(StreamsConfig.producerPrefix("transaction.timeout.ms"), 60000);

Integrating External Stream Processors

Kafka Streams and ksqlDB cover most in-ecosystem processing, but many organizations run hybrid architectures that bring in an external engine. Apache Flink is the common choice for advanced analytics, complex event processing, batch-streaming unification, and workloads that join Kafka against non-Kafka sources. Flink’s Kafka connector supports both source and sink with exactly-once delivery (sink via a two-phase-commit, transactional producer).

Introduce Flink for concrete reasons: sophisticated event-time watermarking, unified batch and streaming, or integration with sources Kafka Streams cannot reach. The operational footprint grows—Flink runs its own cluster, needs checkpoint/savepoint storage (commonly S3 or HDFS), and brings its own monitoring. For deployment and connector patterns, see Integrating Apache Flink with Kafka for Advanced Analytics.

-- Flink SQL table backed by Kafka, with an event-time watermark
CREATE TABLE kafka_orders (
    order_id STRING,
    amount DECIMAL(10,2),
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'properties.bootstrap.servers' = 'broker1:9092,broker2:9092',
    'format' = 'avro',
    'scan.startup.mode' = 'earliest-offset'
);

Operational Monitoring and Observability

Running stream processing in production demands visibility into throughput, latency, consumer lag, state-store size, error rates, and resource use. Kafka Streams publishes a rich metric set over JMX—thread-, task-, and store-level—which you typically scrape into Prometheus and graph. Export it by attaching the JMX-to-Prometheus agent to the JVM:

# Export Kafka Streams JMX metrics to Prometheus via the javaagent
java -javaagent:/opt/jmx_exporter/jmx_prometheus_javaagent.jar=8080:/etc/jmx_exporter/kafka-streams.yml \
     -jar my-streams-app.jar

The signals worth alerting on: consumer lag past a threshold (the app cannot keep up), state-store restore time after a rebalance, and the rate of dropped late records. Useful metrics here include records-lag-max (consumer-level), the per-thread process-rate and process-latency-avg, and the dropped-records metrics. The Kafka Streams monitoring reference lists the full metric set and the MBean names.

Testing and Debugging

Test in layers: unit tests for individual processors and full topologies, integration tests against an embedded or test-container broker, and canary deploys in a production-like environment. TopologyTestDriver runs a topology deterministically with no broker, feeding records in and asserting outputs—fast feedback during development. In production, structured logging with trace IDs, sampling, and interactive queries against live state stores are your debugging tools. For a full methodology, see Testing and Debugging Kafka Streams Applications.

// Unit test using TopologyTestDriver
@Test
public void testOrderEnrichment() {
    TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);
    TestInputTopic<String, Order> inputTopic = testDriver
        .createInputTopic("orders", Serdes.String().serializer(),
            orderSerde.serializer());
    TestOutputTopic<String, EnrichedOrder> outputTopic = testDriver
        .createOutputTopic("enriched-orders", Serdes.String().deserializer(),
            enrichedOrderSerde.deserializer());

    inputTopic.pipeInput("order-1", new Order(100.0));
    assertEquals(1, outputTopic.getQueueSize());
}

Capacity Planning and Performance Tuning

Capacity planning ties together input throughput, per-record processing cost, state size, and infrastructure. The binding relationship is simple: average processing time per record must stay below the inter-arrival time, or lag grows without bound. Horizontal scaling adds stream threads (num.stream.threads) and instances, and scales roughly linearly—up to the partition count of the input topics, which is the hard ceiling on parallelism. If you expect to scale beyond your current instance count, over-partition the input topics up front, because changing partition count later breaks key-to-partition assignment and corrupts stateful results.

State-store size is frequently the constraint. Each stateful task runs its own RocksDB instance, and total disk is the sum across all tasks on a host plus headroom for compaction and restoration. RocksDB tuning—block cache, write buffer size, compression—materially affects read/write performance, but these are set in Java through a RocksDBConfigSetter, not as Streams properties. The RocksDB tuning guide covers the parameters.

// Tune RocksDB via a RocksDBConfigSetter, wired in with
// rocksdb.config.setter=com.example.CustomRocksDBConfig
public class CustomRocksDBConfig implements RocksDBConfigSetter {
    @Override
    public void setConfig(String storeName, Options options,
                          Map<String, Object> configs) {
        options.setCompressionType(CompressionType.LZ4_COMPRESSION);
        options.setWriteBufferSize(64 * 1024 * 1024);   // 64 MiB
        BlockBasedTableConfig table = new BlockBasedTableConfig();
        table.setBlockCacheSize(256 * 1024 * 1024);     // 256 MiB
        options.setTableFormatConfig(table);
    }

    @Override
    public void close(String storeName, Options options) { }
}

Rebalance Tuning and Availability

Rebalances are inherent to distributed stream processing—triggered by instance failures, scaling, or partition changes. A rebalance can suspend processing while tasks move and state restores, causing latency spikes. Kafka Streams uses incremental cooperative rebalancing (KIP-429, the default since 2.4): instead of the old “stop the world, revoke everything” protocol, members keep the partitions they retain and only the ones that must move are revoked, which shortens rolling-bounce and scaling rebalances for heavy stateful apps. Configuring standby replicas (num.standby.replicas) keeps warm copies of state so failover does not pay full restore cost, and acceptable.recovery.lag controls how caught-up a standby must be before it can take over as active without a blocking restore.

# Rebalance and availability tuning for production
session.timeout.ms=45000
heartbeat.interval.ms=3000
max.poll.interval.ms=300000
num.standby.replicas=1
acceptable.recovery.lag=10000

Automation and Infrastructure as Code

Operating stream processing at scale means treating deployments, configuration, and monitoring as code: version-controlled config, automated pipelines, immutable artifacts. Containers and Kubernetes are the common substrate, giving you rolling updates, health checks, and resource isolation. Run stateful Streams apps as a StatefulSet (not a Deployment) with per-pod PersistentVolumeClaims, so each instance keeps stable identity and reattaches its own state volume across restarts—avoiding a full changelog restore on every rollout.

# StatefulSet for a stateful Kafka Streams application
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: order-enrichment
spec:
  serviceName: order-enrichment
  replicas: 3
  selector:
    matchLabels:
      app: order-enrichment
  template:
    metadata:
      labels:
        app: order-enrichment
    spec:
      containers:
      - name: streams-app
        image: registry.example.com/order-enrichment:1.2.0
        resources:
          requests:
            memory: "4Gi"
            cpu: "2"
          limits:
            memory: "8Gi"
            cpu: "4"
        volumeMounts:
        - name: state-store
          mountPath: /data/kafka-streams
        env:
        - name: KAFKA_BOOTSTRAP_SERVERS
          value: "broker1:9092,broker2:9092"
  volumeClaimTemplates:
  - metadata:
      name: state-store
    spec:
      accessModes: ["ReadWriteOnce"]
      resources:
        requests:
          storage: 50Gi

Automated recovery should handle the predictable failures: corrupt or stale local state, poison-pill records that fail repeatedly, and partition reassignment after broker loss. The kafka-streams-application-reset tool resets an application’s offsets and deletes its internal topics so it can reprocess from a clean slate—but it is destructive and must be run carefully in production. The default run executes immediately; pass --dry-run first to preview. Note the flag names: --bootstrap-server is singular (--bootstrap-servers exists but is deprecated), and the tool uses --input-topics plus --internal-topics (the internal changelog/repartition topics, a subset of those it would delete by default). There is no --execute flag. Stop all application instances before running it. See the Application Reset Tool documentation.

# Preview first, then drop --dry-run to execute the reset
kafka-streams-application-reset \
    --application-id order-enrichment-v1 \
    --bootstrap-server broker1:9092 \
    --input-topics orders,customers \
    --dry-run

The reset tool only resets input-topic offsets and deletes internal topics; it does not delete the local RocksDB state on each instance. After resetting, also clean local state on each host (e.g. KafkaStreams.cleanUp() on startup, or remove state.dir) so stale local data does not survive the reset.

Stream processing on Kafka is a mature, well-understood paradigm, but operating it well is a specific skill: sizing and restoring local state, choosing partition keys that don’t skew, getting exactly-once and windowing semantics right, and keeping rebalances cheap. Get those right and the payoff is a real-time data platform that is scalable, resilient, and debuggable. The deep-dive pages linked throughout cover each of these areas in operational detail.