60-Day Kafka 4 Learning Plan · Week 5 — Day 31 of 60

60-Day Kafka 4 Learning Plan · Week 5 — Kafka Streams Sources: Kafka: The Definitive Guide Ch.11 · kafka.apache.org/documentation/streams/developer-guide/dsl-api#windowing

Goal

Understand the four core Kafka Streams window types (including sliding, referenced but not shown in the original material), when each applies, how event-time vs processing-time timestamps affect windowing, how to suppress intermediate results, how to size window state store retention, and how to handle late-arriving events with grace periods.

1. Tumbling windows — fixed, non-overlapping

Each event belongs to exactly one window. Windows are fixed-size and tile the timeline with no overlap.

time →   ●   ●   ●       ●   ●       ●   ●   ●
         [──── W1 ────]  [── W2 ──]  [──── W3 ────]
         [0–5 min]       [5–10 min]  [10–15 min]

// Orders per 5-minute tumbling window
KTable<Windowed<String>, Long> counts = orders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count(Materialized.as("order-counts-per-5min"));

Use cases: Hourly revenue report, per-minute order count, daily active user count.

2. Hopping windows — overlapping, sliding start

Each event can belong to multiple windows. Defined by a size and an advanceBy (the hop interval). When advanceBy < size, windows overlap.

time →   ●   ●   ●   ●   ●       ●   ●
         [──────── W1 ────────]
               [──────── W2 ────────]
         [       0–10 min         ]
               [       5–15 min        ]

Events between min 5–10 appear in both W1 and W2.

// 10-minute window that refreshes every 5 minutes
KTable<Windowed<String>, Long> rolling = orders
    .groupByKey()
    .windowedBy(
        TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(10))
                   .advanceBy(Duration.ofMinutes(5))
    )
    .count();

Use cases: Moving averages, trending content (what’s popular in the last N minutes), rolling fraud scores.

Note: each event is counted size/advance times on average — increases storage and compute proportionally.

3. Session windows — activity-driven, variable size

Groups events by inactivity gap. A session window closes when no event arrives within the gap duration. Window size is variable — determined by how long activity lasts.

time →   ● ● ●    [── 8 min gap ──]    ● ●    [── 6 min gap ──]    ● ● ●
         [S1: 3]                        [S2:2]                       [S3:3]

// Group user clicks into sessions — 5-minute inactivity closes the session
KTable<Windowed<String>, Long> sessions = clicks
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5)))
    .count();

Use cases: User session analytics, bot detection (unusually long/short sessions), click-stream grouping, IoT device activity windows.

Key behavior: When two sessions merge (a new event arrives within the gap of an existing session), Kafka Streams emits a tombstone for the old session and a new record for the merged session.

4. Grace periods — handling late events

In real systems, events arrive late due to network delays, mobile app buffering, or clock skew. Without a grace period, late events are silently dropped.

// Accept late events up to 1 minute after the 5-minute window closes
KTable<Windowed<String>, Long> counts = orders
    .groupByKey()
    .windowedBy(
        TimeWindows.ofSizeAndGrace(
            Duration.ofMinutes(5),   // window size
            Duration.ofMinutes(1)    // grace period
        )
    )
    .count();

Grace period behaviour

⚠️ ofSizeWithNoGrace() drops late events immediately — only safe when you fully control all producers and clocks are tightly synchronised.

⚠️ ofSizeAndGrace() emits multiple results per window as late events arrive — downstream consumers must handle repeated updates for the same window key.

5. Windowed key — reading results

All windowed operations produce a KTable<Windowed<K>, V>. The Windowed<K> key wraps the original key with window boundaries:

// Convert windowed KTable to stream for downstream processing
counts.toStream()
    .foreach((windowedKey, count) -> {
        String userId     = windowedKey.key();
        long windowStart  = windowedKey.window().startTime().toEpochMilli();
        long windowEnd    = windowedKey.window().endTime().toEpochMilli();
        log.info("user={} window=[{},{}] count={}",
            userId, windowStart, windowEnd, count);
    });

When writing to a topic, use a WindowedSerdes to serialize the windowed key:

counts.toStream()
    .to("labs.order-counts",
        Produced.with(
            WindowedSerdes.timeWindowedSerdeFrom(String.class, 5 * 60 * 1000L),
            Serdes.Long()
        ));

6. Window types at a glance

7. Sliding windows — the type the table mentions but doesn’t show

Sliding windows are easy to confuse with hopping windows, but they’re conceptually different: instead of a fixed grid of windows that events fall into, a sliding window is created per event pair — every time a new record arrives, Kafka Streams evaluates a window of the configured size ending at that record’s timestamp.

time →   ●        ●     ●           ●
         [── W(ending at e1) ──]
              [── W(ending at e2) ──]
                    [── W(ending at e3) ──]

// Sliding window: for each new event, look back 5 minutes from its timestamp
KTable<Windowed<String>, Long> slidingCounts = orders
    .groupByKey()
    .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)))
    .count();

Sliding vs hopping — the practical distinction:

Why this distinction matters for the “real-time fraud alerts” use case in the table: a hopping window only refreshes its result every advanceBy interval, meaning a fraud pattern that completes between refreshes waits up to that interval to be detected. A sliding window re-evaluates on every new event, catching the pattern the moment enough evidence exists — at the cost of significantly more computation per event, since every event can trigger a new window evaluation rather than just landing in an existing bucket.

8. Event-time vs processing-time — timestamp extractors

Everything above assumes Kafka Streams knows the “right” timestamp for windowing purposes — but which timestamp it actually uses is a configuration choice with real consequences, not an automatic given.

public class OrderEventTimeExtractor implements TimestampExtractor {
    @Override
    public long extract(ConsumerRecord<Object, Object> record, long partitionTime) {
        OrderEvent event = (OrderEvent) record.value();
        return event.getCreatedAt().toEpochMilli(); // use the business event time, not broker receipt time
    }
}

spring:
  kafka:
    streams:
      properties:
        default.timestamp.extractor: com.labs.streams.OrderEventTimeExtractor

Default behavior warning: without a custom extractor, Kafka Streams uses the record’s Kafka-level timestamp (log append time by default, unless the producer explicitly set CreateTime) — for most real business windowing (revenue per hour, sessions per user), you want a custom extractor pulling the actual event timestamp from your payload, not whatever timestamp Kafka happened to stamp the record with. Getting this wrong doesn’t cause an error — it silently produces windows keyed to the wrong notion of “time,” which is a very easy mistake to ship without noticing in testing (where processing time and event time are usually identical anyway).

9. suppress() — emitting only final window results

By default, a windowed KTable emits an update every time the aggregate changes within the window (plus during any grace period) — useful for live dashboards, but noisy for anything that only cares about the final, settled result per window.

KTable<Windowed<String>, Long> finalCountsOnly = orders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1)))
    .count()
    .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()));

What this changes: instead of downstream receiving one update per contributing event (which, per §4, can be many per window when grace periods are in play), it receives exactly one result per window — emitted only once the window has fully closed (size + grace period elapsed).

Trade-off: suppress() buffers state in memory (or spills to RocksDB with BufferConfig.maxBytes(...) instead of .unbounded()) until the window closes — for a large number of concurrent open windows (e.g. many active user sessions), this is a real memory/disk sizing consideration, not a free operator. Use .unbounded() only when you’re confident about the number of concurrently open windows; use a bounded buffer config in production for anything with unpredictable cardinality, to fail predictably (dropping or spilling) rather than risk unbounded memory growth.

10. Window retention & state store sizing

Windowed aggregations use a specialized RocksDB-backed state store (Day 29 §8) that must retain enough history to answer late-arriving updates and out-of-order records — this retention window is a real disk-sizing input, distinct from the window size itself.

.windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1))
    .grace(Duration.ofMinutes(1)))
// Window store retention defaults to (window size + grace period), but can be set explicitly:
.count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("counts")
    .withRetention(Duration.ofHours(6)))  // how long window results stay queryable, beyond just grace

Two different “how long” numbers to keep straight: the grace period (§4) controls how long late events are still accepted into a window’s computation. Retention controls how long the computed window results remain queryable in the state store (and its changelog) after the window has closed — useful for interactive queries (Day 32) that need to look back further than the grace period alone would suggest. Retention must be ≥ grace period; setting it too low means old window results disappear from queryability sooner than you might expect, even though the underlying events were processed correctly.

11. Testing windowed aggregations

TopologyTestDriver (Day 29 §11, Day 30 §10) needs explicit control over simulated time to test window boundaries — this is where advanceWallClockTime becomes essential.

@Test
void tumblingWindowClosesAfterFiveMinutes() {
    Instant start = Instant.parse("2026-07-31T10:00:00Z");
    input.pipeInput("u1", orderEvent(), start);
    input.pipeInput("u1", orderEvent(), start.plusSeconds(60));

    testDriver.advanceWallClockTime(Duration.ofMinutes(6)); // push past window size + grace
    input.pipeInput("u1", orderEvent(), start.plusSeconds(400)); // triggers window-close processing

    var results = outputTopic.readKeyValuesToList();
    assertThat(results).anyMatch(kv -> kv.value == 2L); // the first window settled at count=2
}

Why this test pattern matters: windowing bugs are invisible in a naive test that just pipes records in immediately back-to-back — the window-closing/grace-period logic only exercises correctly when simulated time actually advances past the relevant boundaries. Always test at least one case where a late event arrives within the grace period and one where it arrives after — these are exactly the two paths §4’s warning describes, and they’re easy to get backwards without a time-aware test.

12. Common pitfalls

Using default (broker-assigned) timestamps for business windowing without a custom extractor — silently windows by the wrong notion of time; works fine in testing where the discrepancy doesn’t show up, breaks assumptions in production under real network delay (§8)
Confusing sliding windows with hopping windows — they solve different problems; a hopping window’s fixed refresh cadence can miss time-sensitive patterns that a sliding window would catch immediately (§7)
Using .unbounded() suppress buffering without considering concurrent window cardinality — unpredictable memory growth for use cases with many simultaneously open windows (e.g. per-user sessions at scale) (§9)
Confusing grace period with retention — setting retention equal to grace period (or forgetting to set it explicitly) can make window results disappear from interactive queries sooner than expected, even though grace period handling itself was correct (§10)
Testing windowing logic without advancing simulated time — a test that pipes all records in immediately never actually exercises window-close or late-arrival code paths (§11)

Key Takeaways

Tumbling: fixed non-overlapping — each event belongs to exactly one window
Hopping: overlapping — each event counted in multiple windows (when size > advance), refreshes on a fixed cadence
Session: variable-size — window closes after inactivity gap, groups activity bursts
Sliding: window evaluated per incoming event relative to its own timestamp — higher fidelity than hopping for true real-time detection, at higher computational cost
Grace period: keeps closed window open for late events — use ofSizeAndGrace()
Use a custom TimestampExtractor for real business event-time windowing — the default broker timestamp is rarely what you actually want
suppress(Suppressed.untilWindowCloses(...)) emits only final results instead of every intermediate update — but buffers state, so size the buffer config deliberately
Window store retention (queryability of results) is a separate, larger number from grace period (late-event acceptance) — retention must be ≥ grace
Windowed KTable key is Windowed<K> — includes window start + end timestamps
withNoGrace() drops late events — safe only if you control all producers
Test window logic with advanceWallClockTime — untested time-boundary code is a common source of silent windowing bugs

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next post.

➡️ Day 32: Aggregations — count, reduce, aggregate + state stores

Resources

📘 Kafka: The Definitive Guide — Chapter 11 (Windowing)
🌐 kafka.apache.org — Windowing DSL
🌐 Kafka Streams — Timestamp Extractors

👉 Link to Medium blog