Stateful Stream Processing with Kafka Streams

Stateful stream processing maintains memory and context across events, enabling computations like windowed aggregations, stream-table joins, and session tracking. Kafka Streams’ embedded state store architecture eliminates external database dependencies for most stateful workloads while providing exactly-once semantics. This article covers the architecture, state store types, operational considerations, and production failure modes.

Stateful processing builds on the broader Stream Processing paradigm. Stateless operations like map, filter, and flatMap process each record in isolation. Stateful operations—aggregate, count, reduce, join—retain information about previously seen records. Kafka Streams embeds this state within the application instance using RocksDB as the default storage engine, backed by a changelog topic in Kafka for durability.

Architecture of State in Kafka Streams

Kafka Streams implements stateful processing through state stores: local, embedded key-value stores residing within the stream processing application. When you define a stateful operation, the framework creates and manages these stores, handling sharding, changelogging, and restoration.

Each partition of the source topic maps to a specific task, and each task owns a dedicated instance of the state store. This co-location of compute and state eliminates network round-trips for state access.

The state store lifecycle is coupled to the consumer group protocol. When a task migrates—due to scaling, failure, or rebalancing—the new host replays the changelog topic to reconstruct the state store before resuming processing. Stateful applications therefore require careful capacity planning: the changelog topic must have sufficient retention and replication, and local disk must accommodate the full state shard.

StreamsBuilder builder = new StreamsBuilder();
StoreBuilder<KeyValueStore<String, Long>> storeBuilder =
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("transaction-counts"),
        Serdes.String(),
        Serdes.Long()
    );
builder.addStateStore(storeBuilder);

KStream<String, Transaction> stream = builder.stream("transactions");
// process() takes an org.apache.kafka.streams.processor.api.ProcessorSupplier
// (the pre-3.0 void process(...) overloads are deprecated by KIP-820)
stream.process(StatefulProcessor::new, "transaction-counts");

The choice between the high-level DSL and the Processor API depends on control requirements. The DSL abstracts state management; the Processor API exposes KeyValueStore, WindowStore, and SessionStore interfaces directly. See Kafka Streams DSL vs Processor API: When to Use Each for guidance.

State Store Types and Operational Footprints

Kafka Streams provides three primary state store types with distinct operational characteristics.

Persistent key-value stores (backed by RocksDB) write data to local disk with an in-memory block cache for hot keys. RocksDB configuration—block cache size, compaction style, write buffer size—directly impacts performance. The framework exposes tuning through RocksDBConfigSetter. Note that on the RocksDB JNI versions shipped with current Kafka, block cache size is set by passing an org.rocksdb.Cache to setBlockCache(...) (the older setBlockCacheSize(long) shortcut has been removed). Implementing close() is mandatory (KIP-453); leave it empty when the cache is shared across stores:

public class CustomRocksDBConfig implements RocksDBConfigSetter {

    // Share a single cache across all store instances in this process.
    private static final org.rocksdb.Cache CACHE =
        new org.rocksdb.LRUCache(128 * 1024 * 1024L); // 128 MB

    @Override
    public void setConfig(String storeName, Options options, Map<String, Object> configs) {
        options.setCompactionStyle(CompactionStyle.LEVEL); // org.rocksdb.CompactionStyle
        options.setMaxWriteBufferNumber(3);
        options.setWriteBufferSize(64 * 1024 * 1024L);     // 64 MB

        BlockBasedTableConfig tableConfig =
            (BlockBasedTableConfig) options.tableFormatConfig();
        tableConfig.setBlockCache(CACHE);
        options.setTableFormatConfig(tableConfig);
    }

    @Override
    public void close(String storeName, Options options) {
        // Shared CACHE is closed at process shutdown, not per store.
    }
}

Register the custom setter via the rocksdb.config.setter Kafka Streams property:

rocksdb.config.setter=com.example.CustomRocksDBConfig

In-memory stores avoid disk but lose state on restart unless changelogging is enabled (it is, by default). They suit small, ephemeral state that reconstructs quickly from the changelog. Window stores extend key-value stores with time-based retention, expiring whole segments once they fall outside the retention period. Segment size determines deletion granularity.

The critical metric is state store size per task. A partition with 10 million keys averaging 1 KB each requires roughly 10 GB of disk per task replica, before RocksDB write amplification. With a replication factor of 3 on the changelog topic, total cluster storage multiplies accordingly.

# Inspect state store disk usage per instance (state.dir layout)
du -sh /var/lib/kafka-streams/*/rocksdb/

# Check changelog topic size on the brokers
kafka-log-dirs --bootstrap-server localhost:9092 \
  --describe --topic-list myapp-transaction-counts-changelog

Changelog Topics and Exactly-Once Semantics

Every persistent state store is backed by a changelog topic—a compacted Kafka topic recording state mutations. During normal operation, writes go to both RocksDB and the changelog. On failure, the new task instance replays the changelog to reconstruct state up to the point of failure.

Compaction prevents unbounded changelog growth by retaining only the latest value per key. Compaction is not instantaneous: it runs on the brokers per segment. The relevant log-cleaner controls are topic/broker configs min.cleanable.dirty.ratio and segment.ms; Kafka Streams sets cleanup.policy=compact on changelogs automatically. A changelog that grows faster than the cleaner can reclaim space will eventually exhaust broker disk.

# Kafka Streams application configuration
replication.factor=3
state.dir=/data/kafka-streams
num.standby.replicas=1
processing.guarantee=exactly_once_v2

With processing.guarantee=exactly_once_v2, Kafka Streams uses idempotent producers and a single transaction per task to ensure output topics and changelog topics reflect each input record exactly once across failures. EOS v2 (introduced as exactly_once_beta and renamed in Kafka 3.0; the older exactly_once and exactly_once_beta values were removed in 4.0) requires brokers on 2.5 or later. Tune transaction.timeout.ms and the broker’s transaction.max.timeout.ms to accommodate commit intervals. For details, see Achieving Exactly-Once Semantics in Kafka Streams with Spring Cloud Stream.

Interactive Queries: Exposing State to External Systems

State stores can be queried at runtime through the interactive query API, turning a stream processor into a lightweight distributed read store for real-time aggregations. Each instance queries only its local shards; the framework provides metadata APIs to discover which instance owns a given key.

Use KafkaStreams.queryMetadataForKey(...), which returns a KeyQueryMetadata exposing activeHost() and standbyHosts() (the older metadataForKey is deprecated):

@GetMapping("/count/{key}")
public ResponseEntity<Long> getCount(@PathVariable String key) {
    KeyQueryMetadata metadata = streams.queryMetadataForKey(
        "transaction-counts", key, Serdes.String().serializer());

    if (metadata == null || metadata == KeyQueryMetadata.NOT_AVAILABLE) {
        return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build();
    }
    HostInfo activeHost = metadata.activeHost();
    if (!activeHost.equals(thisHostInfo)) {
        return redirectToHost(activeHost, "/count/" + key);
    }

    ReadOnlyKeyValueStore<String, Long> store = streams.store(
        StoreQueryParameters.fromNameAndType(
            "transaction-counts",
            QueryableStoreTypes.keyValueStore()));

    Long count = store.get(key);
    return count != null ? ResponseEntity.ok(count) : ResponseEntity.notFound().build();
}

This pattern is covered in Querying State Stores with REST APIs in Kafka Streams. Operationally, each queryable instance is a contention point: provision heap for query serving, and implement readiness checks that wait for KafkaStreams.State.RUNNING before routing traffic. Passing StoreQueryParameters.enableStaleStores() lets standby replicas serve reads during rebalances, trading freshness for availability.

Windowed Aggregations and Temporal State Management

Time-windowed state underpins most real-time analytics. Kafka Streams supports tumbling, hopping, sliding, and session windows, each with different retention requirements. Windowed stores segment data by time; retention determines how long old windows remain queryable. The grace period admits late-arriving records after the window end.

KStream<String, Transaction> stream = builder.stream("transactions");
TimeWindows windows = TimeWindows
    .ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30));

KTable<Windowed<String>, Long> counts = stream
    .groupByKey()
    .windowedBy(windows)
    .count(Materialized.as("windowed-counts"));

Align window retention with business requirements and storage capacity. A 24-hour hopping window advancing every minute keeps roughly 1,440 overlapping windows live per key; with millions of keys, the store can grow to hundreds of gigabytes. Window-store retention (Materialized.withRetention(...), default 1 day, and at least window size + grace) controls when segments are physically deleted: too high wastes disk, too low risks returning empty results for queries near the retention boundary.

For teams using ksqlDB, the operational model differs—ksqlDB manages state stores internally and exposes them through pull queries. See Windowed Aggregations and Joins in ksqlDB for a comparison.

Standby Replicas and High Availability

Stateful applications are exposed to rebalancing latency because a migrating task must replay its changelog before processing resumes. For large stores—hundreds of gigabytes—replay can take minutes. Standby replicas maintain warm copies of state on other instances; the StreamsPartitionAssignor prefers to reassign a task to an instance that already holds a standby, shrinking or eliminating replay time.

num.standby.replicas=2

The cost is proportional: each standby adds a full copy of state storage plus the network and disk to keep it updated. Balance recovery time objectives against infrastructure cost.

Monitoring standby freshness matters—a standby that falls behind under network or disk pressure will not deliver fast failover. Restoration and standby-update progress is exposed through the KIP-869 metrics. Thread-level metrics live under stream-state-updater-metrics (e.g. active-restoring-tasks, standby-updating-tasks, restore-records-rate, restore-call-rate); task-level active-restore progress is restore-remaining-records-total under stream-task-metrics. KIP-988 adds a programmatic StandbyUpdateListener callback for tracking per-store standby lag in application code.

# Read thread-level restore throughput over JMX (Jolokia agent required).
# Task MBeans are kafka.streams:type=stream-task-metrics,thread-id=*,task-id=*
curl -s 'http://localhost:7071/jolokia/read/kafka.streams:type=stream-state-updater-metrics,thread-id=*/restore-records-rate'

Operational Hardening: Disk, Memory, and Rebalancing

Running stateful Kafka Streams in production demands attention to physical resources.

Disk provisioning: RocksDB write amplification typically ranges from 2x to 10x depending on compaction strategy and workload. Provision IOPS and throughput for worst-case amplification, not logical write rate. Use SSDs; spinning disks add latency that compounds during changelog replay and compaction. Point state.dir at dedicated volumes to avoid contention.

mkfs.xfs -f /dev/nvme1n1
mount -o noatime,nodiratime /dev/nvme1n1 /data/kafka-streams

Memory management: RocksDB uses off-heap memory for block cache and write buffers, outside the JVM heap. Size the JVM heap for application objects while reserving system memory for RocksDB and the OS page cache. Bound RocksDB memory explicitly with a shared LRUCache and WriteBufferManager in RocksDBConfigSetter; otherwise per-store buffers multiply with task count and can OOM the host.

Rebalancing: Kafka Streams enables cooperative (incremental) rebalancing by default since 2.4, moving only the tasks that must move and avoiding unnecessary changelog replays. This is handled by the built-in StreamsPartitionAssignor—do not set partition.assignment.strategy to CooperativeStickyAssignor or any other consumer assignor for a Streams app; Streams installs its own assignor and overriding it is unsupported and breaks task assignment.

Monitoring and alerting: useful client- and task-level metrics (all under the kafka.streams JMX domain) include:

Alert when restoration exceeds your recovery time objective, and when a client shows alive threads but zero progress (a stuck rebalance). The Prometheus rule below is illustrative—the exact series name depends on your JMX exporter’s MBean-to-metric mapping, not a canonical Kafka name:

groups:
  - name: kafka_streams
    rules:
      - alert: HighRestorationBacklog
        # Series name produced by your jmx_exporter mapping of
        # stream-task-metrics restore-remaining-records-total
        expr: kafka_streams_stream_task_metrics_restore_remaining_records_total > 1000000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Task {{ $labels.task_id }} has a large restore backlog"

Real-World Failure Modes

Production stateful processing reveals failure modes invisible in development.

Large-state rebalance storm: when an instance with a multi-gigabyte store fails, the rebalance triggers changelog replay on the new owner. If replay saturates network or disk, other instances can time out and cascade. Mitigation: provision standbys, cap restore batch sizes with restore.consumer.max.poll.records (a valid restore-consumer override per KIP-276), and set max.poll.interval.ms high enough to cover replay.

restore.consumer.max.poll.records=100
max.poll.interval.ms=600000

RocksDB write stalls: under heavy write load, RocksDB throttles or stalls writes to let compaction catch up, showing up as increased processing latency and consumer-group timeouts. Mitigation: tune max_write_buffer_number and level0_slowdown_writes_trigger via RocksDBConfigSetter, and scale out to lower per-instance write throughput.

Changelog truncation/divergence: if a changelog topic is deleted, truncated, or its records expire below the store’s checkpointed offset, restoration hits an InvalidOffsetException. Kafka Streams marks the task corrupted (TaskCorruptedException), wipes only the affected store, and re-bootstraps it from the changelog. Mitigation: treat changelog topics as internal—never manage them manually—and restrict access with ACLs.

For a concrete application of these patterns, see Building a Real-Time Fraud Detection System with Kafka Streams.

The maturity of a stateful Kafka Streams deployment is measured by how gracefully it degrades under stress. State stores, changelog topics, and standby replicas form a distributed state-management system that deserves the same operational rigor as any database. Operated correctly, it delivers exactly-once processing, local in-memory/SSD-speed state access, and fast failover within the Kafka ecosystem.