Windowing #

When we analyze real-time data streams, one of the most fundamental questions we must answer is: “How many events happened within a certain time period?” We might want to count website button clicks over the last 5 minutes, monitor hourly bank transfer volumes to detect money laundering, or track user activity in one active interaction session. In Apache Kafka Streams, these temporal analysis needs are facilitated by the Windowing feature. Windowing allows us to divide endless event streams into statefully managed time segment chunks. Through this article, we’ll deeply dissect the four time window types (Tumbling, Hopping, Sliding, and Session), Late-Arriving Data handling techniques using the Grace Period, and how to optimize aggregation result output using the Suppress API.

Basic Windowing Concepts and the Role of Time #

In distributed stream processing, the definition of “time” itself is very complex. Before using Windowing, we must understand the three time concepts supported by Kafka Streams:

  • Event Time: The time when the event actually happened in the real world or on the client device. This time is recorded directly in the record metadata (timestamp) by the producer when data is sent. This is the most important and most accurate time type for windowing aggregations.
  • Ingestion Time: The time when the event was received and written into the Kafka broker log.
  • Processing Time: The local time when the processing thread on the Kafka Streams application instance executes that record. This time type is very inconsistent because it’s influenced by network latency or queue lag.

Kafka Streams processes Windowing based on Event Time by default. This guarantees that even with very long network delays, late-arriving data is still grouped into the correct time window based on its original creation time, not when it arrives at the processing server.

The Four Main Time Window Types #

Kafka Streams provides four time window division models, each designed to solve specific business needs:

1. Tumbling Windows (Fixed Non-Overlapping Time Windows) #

Tumbling Windows divide data streams into fixed-length time segments that connect to each other without overlapping.

  • Characteristics: If we define a 5-minute Tumbling Window, time windows form from 00:00 - 00:05, 00:05 - 00:10, 00:10 - 00:15, and so on. Every event only enters exactly one time window.
  • Use Case: Structured periodic reports, like hourly sales count charts or total bandwidth consumption per 30 minutes.

2. Hopping Windows (Fixed Time Windows with Overlap) #

Hopping Windows have fixed time widths, but hop forward based on a step interval duration smaller than the window size.

  • Characteristics: If we define a Hopping Window with a 5-minute duration and a 1-minute advance interval, we get overlapping windows: 00:00 - 00:05, 00:01 - 00:06, 00:02 - 00:07, and so on. A single event happening at 00:03 enters 4 different time windows in parallel.
  • Use Case: Calculating moving averages, like calculating machine temperatures over the last hour, updated every 5 minutes.

3. Sliding Windows (Dynamic Time Windows Based on Event Arrival) #

Sliding Windows aren’t statically aligned with wall-clock timelines. These windows are dynamically created only when new events enter the system.

  • Characteristics: Time windows are drawn backward for a specific duration (e.g., 5 minutes) from the timestamp of the newly received event. Time windows only cover events happening within the 5-minute boundary before that latest event.
  • Use Case: Fraud detection criteria, like checking whether a user made more than 10 transactions within the last 30 seconds at any time.

4. Session Windows (Time Windows Based on Inactivity Periods) #

Session Windows group tightly occurring events based on individual user activity.

  • Characteristics: These windows have no fixed time size. They’re determined by an inactivity gap boundary. If a user keeps clicking with intervals under 15 minutes, all those clicks are merged into one same Session Window. Once the user stops clicking for more than 15 minutes, that window closes. The user’s next click starts a new Session Window.
  • Use Case: Web user behavior analysis, game session duration tracking, or customer shopping sessions in mobile applications.
EACH WINDOW IN DETAIL:
-------------------------------------------------------------------
Tumbling (Size: 5s):
  [0s-5s]   [5s-10s]  [10s-15s]

Hopping (Size: 5s, Advance: 2s):
  [0s-5s]
    [2s-7s]
      [4s-9s]

Session (Inactivity Gap: 5s):
  ●──●──● (Event) ──── [10s Gap] ──── ●──● (Event)
  [==== Session 1 ====]               [== Session 2 ==]
-------------------------------------------------------------------

Late Data Handling: Grace Period and Late Arrivals #

In distributed systems, late arrivals are inevitable because of frequently interrupted mobile internet connections. For example, an IoT sensor device sends temperature data at 08:00. However, because of signal loss in a tunnel, that data only arrives at our Kafka broker at 08:15.

By default, if we don’t configure a tolerance limit, Kafka Streams processes that record. However, if the state store for the 08:00 window was already deleted from local RocksDB because it’s considered too long passed, that late data is silently discarded.

Solution: Grace Period #

To solve this problem, we must explicitly determine the grace() configuration when defining windowing.

  • How It Works: The Grace Period determines how long already-passed time windows are allowed to stay open in RocksDB memory to accommodate late-arriving data.
  • Lifecycle: If we create a 5-minute Tumbling Window with a 2-minute Grace Period, the 08:00 - 08:05 time window stays open in local RocksDB until the system Stream Time reaches 08:07. Any late data timestamped between 08:00 - 08:05 arriving before 08:07 is still counted and inserted into that window’s aggregation. After 08:07 passes, that window is permanently closed, and subsequent late data for that window is discarded.

The Suppress API: Reducing Intermediate Update Bursts #

By default, every time a new record enters an active window, Kafka Streams immediately emits the latest aggregation result to the output topic (downstream). If we count transactions per hour, and there are 1,000 transactions in one hour for one user, KStreams emits 1,000 update records to the output topic.

This triggers severe performance problems (database write amplification) if our downstream system is a relational database or slow third-party API.

Solution: The Suppress API #

To suppress these intermediate update bursts, we use the suppress() operator.

  • How It Works: Suppress instructs Kafka Streams to hold all intermediate updates in internal memory buffers.
  • Final Result: KStreams only emits one single record (the final calculation result) downstream right when that time window is permanently closed (after passing the window duration + Grace Period).

Implementation Code: Anti-Pattern vs Stateful Windowing Solutions #

Let’s study windowing implementation using the Java SDK, comparing anti-pattern approaches with safe managed solutions.

Use Case #

We want to monitor user clicks on e-commerce pages (page-click-events) and count total clicks per user every 5 minutes. We want to accommodate late data up to 1 minute, and only send the final calculation result to the final-click-counts output topic.

Anti-Pattern: Manual Windowing Using Java Threads / Timers #

Trying to manually cut time flows in memory using Java scheduler threads (ScheduledExecutorService) or timer loops is an anti-pattern damaging data consistency during crashes.

// ANTI-PATTERN: Doing manual windowing calculations using Timer / Java Scheduled Threads.
// ✗ Not integrated with Event Time, prone to data loss on restarts, and doesn't support distributed Grace Periods.
public class ManualWindowProcessor {
    private static final Map<String, Integer> clickCounts = new ConcurrentHashMap<>();

    public static void processStream(KStream<String, String> stream) {
        // ✗ DANGER: Time cutting based on local wall-clock system time (Processing Time)
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
        executor.scheduleAtFixedRate(() -> {
            // Unsafely emitting periodic results to external databases
            clickCounts.forEach((userId, count) -> {
                log.info("User {} clicked {} times in the last interval", userId, count);
            });
            clickCounts.clear(); // Clear the manual cache (prone to thread races!)
        }, 0, 5, TimeUnit.MINUTES);

        stream.foreach((userId, event) -> {
            clickCounts.merge(userId, 1, Integer::sum);
        });
    }
}

Practical Solution: Aggregation with Tumbling Windows and the Suppress API #

Below is the correct, recommended Java SDK code. We configure a 5-minute Tumbling Window, set a 1-minute Grace Period, and apply the Suppress API to ensure only the final result is sent.

// CORRECT: Using KGroupedStream, TimeWindows with a Grace Period, and the Suppress API.
// ✓ Guaranteeing Event Time-based accuracy, safe from crash failures, and very bandwidth efficient.
public class ResilientWindowAggregation {
    
    public static void buildTopology(StreamsBuilder builder) {
        KStream<String, String> clickStream = builder.stream(
            "page-click-events",
            Consumed.with(Serdes.String(), Serdes.String())
        );

        // Step 1: Group data by User ID (Key)
        KGroupedStream<String, String> groupedClicks = clickStream.groupByKey(
            Grouped.with(Serdes.String(), Serdes.String())
        );

        // Step 2: Define a 5-minute Tumbling Window with a 1-minute Grace Period
        // ✓ Forcing the parser to retain window status in local RocksDB
        TimeWindows tumblingWindow = TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))
            .grace(Duration.ofMinutes(1)); // Late data tolerance up to 1 minute

        // Step 3: Statefully aggregate data into the time windows
        KTable<Windowed<String>, Long> windowedCounts = groupedClicks
            .windowedBy(tumblingWindow)
            .count(
                Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("click-count-store")
                    .withKeySerde(Serdes.String())
                    .withValueSerde(Serdes.Long())
            );

        // Step 4: Apply the Suppress API to hold intermediate updates
        // ✓ Only emitting the final record to the destination topic when the window closes (after minute 6)
        windowedCounts
            .suppress(Suppressed.untilWindowCloses(
                Suppressed.BufferConfig.maxRecords(10000) // Safety memory buffer limit
                    .shutDownWhenFull() // Safety if the buffer explodes
            ))
            .toStream()
            // Converting the Windowed<String> key to a regular String for downstream readability
            .selectKey((windowedKey, count) -> String.format("%s@%d-%d", 
                windowedKey.key(), 
                windowedKey.window().start(), 
                windowedKey.window().end()
            ))
            .to(
                "final-click-counts",
                Produced.with(Serdes.String(), Serdes.Long())
            );
    }
}

Cleaning Up Expired Window State Stores #

Window State Stores in RocksDB must not bloat forever. To prevent disk space exhaustion on local servers, Kafka Streams automatically manages the gradual cleanup of passed window data.

  • Retention Time: By default, the window state store retention time is the window duration plus the Grace Period multiplied by two ((Window Size + Grace Period) * 2).
  • Physical Segment Division: Under the hood, RocksDB divides the persistent database into several separate physical segment files based on time. When a segment is fully filled with window data that has passed the retention period, the RocksDB segment containing the related time windows is permanently deleted from local disk at once.
  • This physical segment-based deletion is far more efficient than logically deleting rows one by one, because it doesn’t trigger data fragmentation in RocksDB. Late data arriving after the retention time expires isn’t processed and is discarded.

Summary #

  • Event Time — The main time parameter used in Kafka Streams Windowing to group records based on when events happened on the client side.
  • Tumbling Windows — Fixed-duration time windows connecting linearly without overlapping.
  • Hopping Windows — Fixed-duration time windows advancing forward with smaller shift intervals, causing overlapping.
  • Session Windows — Dynamic time windows without fixed sizes determined by user data inactivity periods (inactivity gaps).
  • Grace Period — The additional wait tolerance letting already-passed time windows stay open in RocksDB to accommodate late data.
  • Suppress API — The record burst cutting operator holding intermediate updates in memory and only emitting final results after windows are permanently closed.
  • Retention Time — The window data retention limit on local RocksDB disk before permanent deletion to prevent storage space exhaustion.

← Previous: Fault Tolerance Next: Join Stream and Table →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact