Kafka Streams: Building Real-Time Stream Processing Applications

Kafka Streams is a library, not a cluster. Your application reads from Kafka, processes, and writes back to Kafka. There is no separate processing cluster to operate, no resource allocation to negotiate, no additional operational overhead.

Most stream processing frameworks (Flink, Spark Streaming) run on dedicated clusters. Kafka Streams runs on application infrastructure you already manage. That is the core difference.

Stream Processing Basics

A stream is an unbounded sequence of records. A stream processor reads from input streams, applies transformations, and writes to output streams.

flowchart LR
    Input[Input Topic] --> KS[Kafka Streams App]
    KS --> Output1[Output Topic 1]
    KS --> Output2[Output Topic 2]

Kafka Streams programs define processing logic using the Streams DSL or the Processor API. The DSL provides functional-style operations on streams. The Processor API gives you lower-level control when you need it.

The Streams DSL

The Streams DSL exposes KStream and KTable abstractions.

KStream: An unbounded sequence of record events. Each event is an independent occurrence. The same key can appear multiple times.

KTable: A mutable view of a changelog stream. Updates to a key replace the previous value. A KTable is backed by a state store that maintains the latest value for each key.

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Grouped;

StreamsBuilder builder = new StreamsBuilder();

// KStream: unbounded sequence of page view events
KStream<String, PageViewEvent> pageViews = builder.stream("page-views");

// KTable: latest customer info per customer ID
KTable<String, CustomerInfo> customers = builder.table("customer-info",
    Materialized.as("customer-info-store"));

// Join page views to customer info
KStream<String, EnrichedPageView> enrichedPageViews = pageViews
    .filter((userId, event) -> event.getDuration() > 10)  // filter long sessions
    .join(customers, (event, info) -> new EnrichedPageView(event, info));

// Aggregate page views per user
KTable<String, Long> viewCounts = pageViews
    .groupBy((userId, event) -> userId)
    .count(Materialized.as("view-counts-store"));

// Write to output topics
enrichedPageViews.to("enriched-page-views");
viewCounts.toStream().to("view-counts");

Stateful Operations

Kafka Streams keeps state in local RocksDB databases by default. Each instance stores the current state for the partitions it handles.

Stateful operations include:

• join: Join two streams or a stream with a table
• aggregate: Maintain running aggregates (count, sum, max)
• reduce: Combine records with the same key

// Maintain a running count per user with session timeout
KStream<String, Event> events = builder.stream("events");

events
    .groupByKey(Grouped.with(Serdes.String(), eventSerde))
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .aggregate(
        () -> new SessionState(0, Instant.MIN),
        (key, event, state) -> state.addEvent(event),
        (key, left, right) -> left.merge(right),
        Materialized.as("session-store")
    )
    .toStream()
    .to("session-output");

The state store is partitioned by key. Each Kafka Streams instance hosts a subset of partitions and thus a subset of state store partitions. When a Kafka Streams instance fails, another instance takes over its partitions and rebuilds its state store from the changelog topic.

Exactly-Once Processing

Kafka Streams provides exactly-once processing within the Kafka ecosystem. A record is processed exactly once even if failures occur.

The mechanism is transactional output. Kafka Streams uses Kafka transactions to coordinate output with offset commits. If a processor crashes after writing output but before committing its offset, the transaction aborts and the output is rolled back.

// Enable exactly-once processing
Properties props = new Properties();
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-exactly-once-app");

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> input = builder.stream("input-topic");
KStream<String, String> output = input.mapValues(value -> process(value));
output.to("output-topic", Produced.with(Serdes.String(), Serdes.String()));

Exactly-once within Kafka means exactly-once relative to Kafka. If your processor reads from Kafka, writes to an external database, and then crashes before committing the Kafka offset, the external database write happened but the offset was not committed. On restart, the record is reprocessed and the external write repeats.

For true end-to-end exactly-once with external systems, you need the transactional outbox pattern. See Exactly-Once Semantics.

Scaling Kafka Streams Applications

Kafka Streams scales by running multiple instances of the same application. Each instance processes a subset of partitions. More instances means more parallelism, up to the number of partitions.

flowchart LR
    subgraph Kafka[Kafka Cluster]
        P0[Partition 0]
        P1[Partition 1]
        P2[Partition 2]
    end
    subgraph Instances[Kafka Streams App]
        I1[Instance 1: P0, P1]
        I2[Instance 2: P2]
    end
    P0 --> I1
    P1 --> I1
    P2 --> I2

Adding instances is handled automatically. When an instance joins, partitions are rebalanced. When an instance leaves, its partitions are reassigned. The state store for each partition is recreated from the changelog.

The scaling ceiling is the number of partitions. If you have 6 partitions, you can have at most 6 Kafka Streams instances processing in parallel. This is one reason partition count matters for Kafka applications.

Interactive Queries

Kafka Streams exposes state stores via interactive queries. Your application can query its local state store directly without going through Kafka.

// Expose state store as a queryable API
ReadOnlyKeyValueStore<String, SessionState> store =
    streams.store(
        QueryableStoreTypes.keyValueStore(),
        StoreQueryParameters.fromNameWithType("session-store", SessionState.class)
    );

SessionState state = store.get("user-123");

This is useful for building real-time applications that need low-latency access to computed state. A sessionization application can serve session data directly from its local state store without querying a database.

Common Mistakes

Storing too much in state stores

State stores are local RocksDB databases. They are designed for incremental state updates, not for storing millions of records per partition. If your state store grows unbounded, your application runs out of disk and crashes.

Always use windowed state stores with appropriate retention. Set grace periods correctly so that late-arriving data does not create infinitely growing windows.

Size thresholds. A state store becomes problematic when it exceeds roughly 1 GB per partition per instance. At that scale RocksDB compaction cycles start to affect processing latency. Above 10 GB per partition you are at serious risk of disk exhaustion during a compaction spike or changelog replay. These are rough figures. The actual safe limit depends on your disk size, available I/O bandwidth, and how much disk you are willing to dedicate to the state store versus the OS and other services on the same host.

Two growth patterns. Unbounded growth comes from one of two causes. The first is key cardinality growth: you have a fixed or slowly growing number of keys but each key’s value blob keeps expanding. This happens with session windows that never close because the gap timeout is set to zero or the grace period is infinite. The second is key count growth: you genuinely have more keys over time, either because your user base is growing or because your event source keeps emitting new entity IDs. Cardinality growth is the more dangerous pattern because it is harder to detect. A single key absorbing gigabytes looks the same as a healthy store in most monitoring tools until you run out of disk.

RocksDB settings to bound size. Configure max_bytes_for_level_base and max_bytes_for_level_multiplier to cap compaction target sizes at each level. Set a TTL on column families to expire old data automatically if your use case supports it. For time-windowed stores configure the retention period on the window manifest so old segments get deleted after the grace period closes. The Kafka Streams setting state.store.max.bytes hard-caps the store size. Writes block or throw when the limit is hit, which is preferable to disk exhaustion.

Early warning metrics. Track kafka_stream_state_store_size_bytes per task. Also watch kafka_stream_state_store_changelog_lag_bytes. A growing changelog lag with a static store size means new keys are being created faster than old ones are being expired, which is the precursor to cardinality explosion. Watch for rocksdb.num_entries_active_mem_table spiking repeatedly, which signals compaction backlog. If you see disk I/O wait climbing on the state store host while the store size metric is flat, compaction is falling behind.

Example: session windows with no gap timeout. Session windows with TimeWindows.ofSizeWithNoGrace(Duration.ofZero()) never close. Each unique key accumulates events indefinitely. After a year of continuous events on even a moderate-traffic key the store holds millions of records under a single key. The fix is to set a gap timeout. A session window with a five-minute gap closes and evicts completed sessions automatically, keeping the store bounded by time rather than by event count.

Example: time windows with proper grace. Time windows naturally cap at their size. A tumbling window of one hour holds at most events_per_hour / unique_keys records per key. Adding a grace period of five minutes allows late arrivals to land in the correct window without reopening it. Once the grace period expires the window closes and RocksDB drops the underlying data during the next compaction cycle. The store size stays proportional to your event rate and key count, not to your total event history.

Not handling out-of-order data

Kafka guarantees ordering within a partition. If your processing logic depends on temporal ordering across partitions, you must handle out-of-order arrivals explicitly.

The most common mistake is assuming that later records on the same partition are always chronologically later. They are usually right, but not always. Producers retry when acknowledgments do not arrive, consumer groups rebalance mid-flight, and transient network issues can reorder events. The same logical event sometimes lands twice with different timestamps, or in the wrong temporal order. Session windows and time-based joins break in ways that are hard to detect because nothing throws an exception — you just get wrong results.

The fix is to rely on event time from the record itself, not processing time. Set up a timestamp extractor that pulls from the event payload, not Kafka’s metadata timestamp. Then configure a grace period that reflects how out-of-order your data typically gets. Events that arrive after the grace period closes get dropped rather than reopening a window, which would otherwise require expensive state rewinding.

For joins, co-partitioning the input topics on the same key removes the cross-partition ordering problem entirely. If you cannot co-partition, use a stream-to-table join instead. The KTable’s changelog semantics naturally handle late arrivals because it overwrites the previous value, keeping the joinable state current regardless of arrival order.

Using global state stores when partitioned would work

Global state stores replicate all data to every Kafka Streams instance. They are convenient but do not scale. Use partitioned state stores whenever possible.

When to Use Kafka Streams

Use Kafka Streams when:

• Your entire pipeline is Kafka-based (sources and sinks are Kafka topics)
• You want to embed processing in your application rather than operate a separate cluster
• You need exactly-once within the Kafka ecosystem
• Your processing is relatively straightforward (filter, map, join, aggregate)

Do not use Kafka Streams when:

• You need complex event processing (CEP) with patterns like sliding windows, sessionization with complex rules
• You need to integrate with multiple heterogeneous sources (Kafka, Kinesis, Pub/Sub, file systems)
• You need advanced windowing (count windows, multiple windows, late triggers)

Trade-off Table: Kafka Streams vs Flink vs Spark Streaming

Aspect	Kafka Streams	Apache Flink	Spark Streaming
Deployment model	Embedded library	Separate cluster	Separate cluster
Operational complexity	Low	High	High
State management	Local RocksDB	Distributed RocksDB	In-memory or tiered
Exactly-once guarantee	Within Kafka only	End-to-end	Micro-batch exactly-once
Latency	Sub-millisecond	Sub-millisecond	~500ms (micro-batch)
Windowing support	Time, session, sliding	Time, session, count, global	Time only (micro-batch)
CEP / pattern detection	Limited	Rich (MATCH_RECOGNIZE)	No
Job recovery	Task restart from checkpoint	Fine-grained restart	Checkpoint restart
Scaling	Partition count ceiling	Rescales dynamically	Partition count ceiling
Kafka dependency	Required	Optional (many connectors)	Optional
Best for	Kafka-native microservices	Complex event processing, large-scale streaming	Batch-oriented streaming, legacy migration

Kafka Streams works best when everything lives in Kafka and you want to keep operations simple. Flink handles complex event patterns and large-scale state. Spark Streaming suits teams moving from batch Spark who want a familiar API.

Production Failure Scenarios

Stateful Kafka Streams applications fail in ways stateless ones do not. These are predictable failure modes — plan for them before they happen.

State store corruption

RocksDB can corrupt after a disk write failure or an unclean shutdown. When the application restarts, it crashes trying to read the corrupted store.

The corruption usually surfaces one of two ways. The application throws a RocksDBException during startup when opening the store, or individual read operations return checksum mismatches during normal processing. Both point to a memtable flush or WAL write that did not complete cleanly. On SSDs, a sudden power loss or unreported write error is the typical cause. On HDDs, silent data corruption from disk surface errors occasionally surfaces in state stores months later.

Mitigation: enable RocksDB checksums (options.verify_checksums = true). This makes RocksDB validate checksums on every read, catching corruption immediately rather than propagating bad state. Monitor disk health using SMART metrics — many hosted environments expose these — and watch for rising RocksDBException rates in your application metrics. If corruption happens, the store rebuilds from the changelog topic. Recovery time is changelog_lag / restore_throughput: a 10M row changelog restoring at 50K rows/sec means about three minutes before the app can serve traffic again. For large state stores, incremental checkpointing reduces replay time.

Disk I/O errors that cause corruption tend to recur. If a specific disk is failing, subsequent restarts may hit corruption again even after rebuild. Route state store I/O to healthy disks, and replace degraded storage before it causes repeated incidents.

Rebalance deadlock

An unclean shutdown (SIGKILL, OOM kill) triggers a rebalance. During rebalance, no instance processes the affected partitions. If the shutdown itself was caused by a processing bug — say, a join that deadlocked on a shared resource — new instances taking over hit the same bug and also crash.

The failure cycle is what makes this dangerous. Consumer A crashes from the bug. During rebalance, Consumer B picks up Consumer A’s partitions and triggers the same bug. Consumer B crashes. Another rebalance starts. The group never stabilizes, and each crash generates a new rebalance event. You see Member [...] has left group and Rebalancing events repeating in logs with no stable interval between them — the signature of a rebalance storm.

The bug that triggers this is usually a shared resource bottleneck under load, not a simple NPE. Common patterns: a join against an external service where connection pool exhaustion causes a timeout, a synchronized block on a shared lock that held during the crash, a rate limiter that hit its ceiling and blocked processing indefinitely. The bug is latent — it only surfaces under the specific concurrency conditions during rebalance when a single instance processes partitions from two consumers.

Mitigation has three layers. First, make processing idempotent so reprocessing is safe even if it runs multiple times. Second, set max.poll.interval.ms high enough to absorb processing spikes, and max.poll.records low enough that each batch finishes well within that window. Third, use exactly_once_v2 so offsets and output commit atomically, which prevents duplicate window state accumulation during the rebalance-and-retry cycle. For the underlying bug itself, eliminate shared resource bottlenecks from your processing code: use connection pools with timeouts, avoid synchronized blocks on hot paths, and make external service calls non-blocking where possible.

Duplicate output from crash during stateful join

A stateful join holds state in a repartition topic and co-partitioned state store. If the instance crashes after writing to the output topic but before committing the offset, the offset rolls back. On restart the join re-executes and produces duplicates.

The exact failure sequence: the join operator reads a record, looks up the matching record in the state store, produces an enriched result to the output topic, and then waits for the consumer poll interval to call commit() before committing the offset. If the instance crashes after the output write succeeds but before the offset commit, the offset is not recorded. On restart, the consumer group coordinator reassigns the partition and a new instance re-reads the same record from the offset. The join executes again. The output topic receives the same enriched result twice.

The window where this happens is max.poll.interval.ms wide. Any crash within that window after an output write but before the next poll triggers the duplicate. Network timeouts, GC pauses that extend past the session timeout, and slow downstream sinks that block send() all extend that window.

Mitigation: exactly_once_v2 guarantees that the output write and the offset commit together or roll back together. When the instance restarts, either the previous output is present and the offset has moved, or the previous output is absent and the offset has not moved — no gap where one is committed and the other is not. Design downstream consumers idempotently as a second line of defense: use primary key upserts, not inserts, so that a duplicate write overwrites the first rather than creating a second row.

Silent data loss from grace period violations

Late-arriving data beyond the grace period drops silently. If your watermark config mixes event time and wall clock time, you may silently lose events with no exception raised.

The worst case is a watermark strategy that blends event time and processing time. If your WatermarkStrategy uses forBoundedOutOfOrderness(Duration.ofSeconds(30)) without explicitly assigning event timestamps, Flink uses the Kafka metadata timestamp — which is processing time, not event time. A 30-second network hiccup causes the watermark to stall 30 seconds behind wall clock. Events that arrived during the hiccup look late by event time and get dropped, not because they were truly late, but because the timestamp extractor was wrong. No exception is raised. The data is simply gone.

Three concrete scenarios cause silent watermark misconfiguration. First, a TimestampAssigner that reads from the wrong field — for example, using the record arrival time instead of the event’s embedded timestamp. Second, mixing sources with different timestamp precisions — Kafka metadata timestamps in milliseconds versus database row timestamps in seconds. Third, a grace period that is too short for your actual out-of-orderness distribution. A 5-second grace period sounds strict until you measure your actual p99 out-of-orderness and find it is 45 seconds during peak load.

Mitigation: always set the TimestampAssigner explicitly and test it with events that have known timestamps. Set the grace period based on measured out-of-orderness distributions, not gut feel. Monitor late-record-dropping per task — non-zero values mean your grace period is not keeping up. Alert on watermark stalling: a frozen watermark means one partition has stopped advancing and late events are piling up and being dropped. Configure withIdleness(Duration) on the watermark strategy so that idle partitions do not stall the overall watermark.

Capacity Estimation

State store sizing

Each state store partition holds keys distributed by your partition count. Work out the per-partition size before production.

// State store size estimation per partition
// Assumptions: 100M total keys, 100 partitions, 500 bytes per value, 2x RocksDB overhead

long totalKeys = 100_000_000L;
int numPartitions = 100;
long keysPerPartition = totalKeys / numPartitions;  // 1M keys per partition

long bytesPerValue = 500L;
long rocksDBOverheadFactor = 2L;  // indexes, bloom filters, tombstones

long stateStoreBytes = keysPerPartition * bytesPerValue * rocksDBOverheadFactor;
// ≈ 1GB per partition per instance

// With 3 instances (co-partitioned), each instance holds ~1GB of state
// Add 50% for changelog backlog during recovery
long totalLocalStorageNeeded = stateStoreBytes * 1.5;  // ~1.5GB per instance

Sizing checklist: divide total keys by partitions to get keys per partition per instance, multiply by average value size and 2x for RocksDB overhead, then add 1.5-2x buffer for the changelog backlog. On disk, RocksDB is typically 1.5-3x its in-memory size. Use the state-store-size metric to validate against your estimate.

Throughput

Kafka Streams scales linearly with partition count until you hit a bottleneck — CPU, network, or disk IO.

Operation	Records/sec per core
Stateless map/filter	50–100K
Stateful join/aggregate	10–30K
Windowed sessionization	5–15K

An 8-core machine hits 400–800K records/sec stateless. Stateful work drops that to 80–240K records/sec. Profile with your actual data shapes — the variance is significant.

Observability Checklist

Monitor these metrics in production:

Throughput: records-consumed-rate, records-produced-rate, bytes-consumed-rate. Sudden drops mean a rebalance is in progress or an upstream producer failed.

State store: state-store-size per task. Growing stores without a corresponding business reason mean missing cleanup or an expanding key set.

Lag: consumer-lag per partition. Lag that grows without bound means the app cannot keep up with incoming data.

Rebalance: commit-latency and rebalance-total-count. Frequent rebalances without instance churn point to a processing deadlock or timeout.

Watermark: watermark-entered-latest-timestamp and late-record-dropping-per-second. Non-zero late drops means your grace period is misconfigured or event time is lagging behind wall clock.

Alert thresholds: rebalance count above 3 per hour outside planned restarts, consumer lag growing for more than 10 minutes, state store size growing more than 10% day-over-day without explanation.

Quick Recap

• Kafka Streams is a library, not a cluster — runs inside your application.
• Size state stores before production. RocksDB grows until it hits disk limits.
• Exactly-once only covers Kafka-to-Kafka. External database writes need the transactional outbox pattern.
• Partition count is your scaling ceiling — pick it with growth in mind.
• Rebalances pause processing. Avoid unclean shutdowns to keep them rare.
• Watch state store size, consumer lag, and rebalance frequency as your primary signals.

Conclusion

Kafka Streams is a powerful library for stream processing applications that integrate tightly with Kafka. It provides exactly-once processing, stateful operations, and elastic scaling within the Kafka ecosystem.

The operational simplicity is the biggest advantage. There is no processing cluster to manage, no resource allocation to tune. The application is just another Kafka consumer group.

The limitation is the Kafka boundary. When your processing must span multiple systems, or when you need advanced windowing features, dedicated stream processing frameworks like Apache Flink become necessary.