ReviseAlgo Logo

Kafka Streams

Stateful Operations — aggregations, joins, windowing

Tumbling, hopping, session windows and state stores (RocksDB).

Introduction

Stateful Operations are transformations in Kafka Streams that require tracking history or storing intermediate states. These include aggregations (e.g. counting events), stream/table joins, and windowing configurations.

Why It Matters

Most stream applications require stateful context. You cannot detect credit card fraud without comparing a transaction to past user patterns, and you cannot compute hourly traffic reports without aggregating counts into set time windows. Senders must master stateful mechanics to handle out-of-order events and state failovers safely.

Real-World Analogy

Think of stateful operations like a receptionist keeping tally marks on a clipboard. If people enter a lobby (events), a stateless receptionist counts each entry individually without memory. A stateful receptionist checks the clipboard, increments a tally by key (e.g. "active guests from 2 PM to 3 PM"), and references profiles to ensure guest records match.

How It Works

Stateful operations utilize local state stores backed by Kafka changelog topics:

  1. Local Store Write: The active processor writes intermediate aggregates to the local RocksDB instance.
  2. Changelog Replicaton: Changes to the local store are simultaneously appended to a dedicated Kafka changelog topic.
  3. Window boundaries: Aggregations are grouped by windows (Tumbling, Hopping, Session) that define the start and end constraints of computed state.

Internal Architecture

Kafka Streams supports four windowing patterns:

  • Tumbling Windows: Fixed-size, non-overlapping windows (e.g., 5-minute blocks).
  • Hopping Windows: Fixed-size, overlapping windows defined by size and an advance interval (e.g., 5-minute window advancing every 1 minute).
  • Sliding Windows: Dynamic windows based on time differences between records.
  • Session Windows: Windows defined by periods of inactivity (inactivity gaps).

Visual Explanation

Below is an ASCII representation of Tumbling vs Hopping vs Session windows: 

[Events Over Time] ───(A)─────(B)──(C)──────────────(D)─────(E)──────>
Tumbling (5m)      |─── Window 1 ───|─── Window 2 ───|─── Window 3 ───|
                      (A, B, C)            ()             (D, E)

Hopping (5m, adv 2m)  |─── Win 1 ───|
                        |─── Win 2 ───|
                          (A, B, C)
Session (Gap 3m)   |── Session 1 ───|                 |── Session 2 ──|
                      (A, B, C)                           (D, E)

Practical Example

Below is a Java snippet showing how to implement a tumbling window count of user clicks:

Common Mistakes

  • Not setting grace periods for windows: If you do not configure a grace period, late-arriving events are immediately dropped instead of being aggregated into their respective historical windows.
  • Omitting materialized state store configurations: If you let stateful operators use defaults without specifying names, Kafka Streams generates random store names, causing state incompatibility bugs during redeployments.

Quick Quiz

Q1: What is the purpose of the Kafka changelog topic associated with stateful aggregations?

To provide durable backups for local RocksDB states, allowing state stores to be reconstructed on failover.

Q2: What is the difference between tumbling and hopping windows?

Tumbling windows are fixed-size and non-overlapping; hopping windows are fixed-size but overlap since they advance in steps smaller than their size.

Scenario-Based Challenge

The Challenge: Detecting fraudulent multi-purchase patterns

You need to raise a fraud alert if a credit card is used 3 or more times within any sliding 1-minute period. How do you construct this aggregation topology?

Solution Tip: Group the incoming transactions stream by credit card number. Apply a sliding window of size 1 minute (SlidingWindows.ofLimit(Duration.ofMinutes(1))). Apply a count operator, and filter the output where the aggregate count is greater than or equal to 3.

Debugging Exercise

Your application logs a high volume of errors: RocksDBException: Failed to open database: ... Too many open files.

The Fix: Check the number of stateful tasks running on the host. Each state store instance opens multiple files. Increase the Linux file descriptor limits (ulimit -n 65536) or optimize RocksDB configuration parameters via a custom RocksDBConfigSetter implementation.

Interview Questions

1. How does Kafka Streams handle late-arriving events in windowed aggregations?

It uses the grace period configuration. If an event's timestamp falls within the window size + grace period, it is aggregated; otherwise, it is permanently discarded.

2. What are standby replicas and how do they reduce recovery time?

Standby replicas are mirror tasks running on other active instances. They replicate the changelog topic asynchronously, allowing instantaneous failovers.

Production Considerations

Always configure num.standby.replicas=1 for critical stateful topologies to guarantee low recovery times during container migrations.