Windowing in Kafka Streams
AdvancedTumbling windows (fixed non-overlapping), hopping windows (overlapping), session windows (activity-based gaps), and sliding windows each model different time-series aggregation needs.
Overview
Windowing in Kafka Streams groups records by time intervals for aggregation. The four window types address different analytical needs. Tumbling windows divide time into fixed, non-overlapping buckets (hourly revenue, daily active users). Hopping windows slide forward by a step smaller than the window size, producing overlapping results (5-minute average CPU every 1 minute). Session windows group records by user activity gaps — a session ends when a user is idle for more than the inactivity gap — useful for user behavior analysis. Kafka Streams uses event time by default (the record timestamp) which correctly handles out-of-order events; late records are processed up to a grace period after the window closes, then discarded.
Tumbling and hopping windows
Tumbling windows for exact hourly/daily buckets; hopping windows for rolling aggregations like 5-minute average over last hour.
StreamsBuilder builder = new StreamsBuilder();
KStream<String, OrderEvent> orders = builder.stream("order-events");
// TUMBLING window — count orders per product per 1-hour bucket
// Windows: [00:00–01:00], [01:00–02:00], [02:00–03:00] ...
KTable<Windowed<String>, Long> hourlyCounts = orders
.groupBy((key, order) -> order.getProductId())
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
// Grace period: accept late events for 5 minutes after window closes
// .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofHours(1), Duration.ofMinutes(5)))
.count(Materialized.as("hourly-product-counts"));
// Read windowed results
hourlyCounts.toStream()
.map((windowedKey, count) -> {
String productId = windowedKey.key();
long windowStart = windowedKey.window().start();
long windowEnd = windowedKey.window().end();
return KeyValue.pair(productId,
String.format("%s: %d orders in %d–%d", productId, count,
windowStart, windowEnd));
})
.to("hourly-product-counts-output");
// HOPPING window — 5-minute sum, updated every 1 minute (overlapping)
// Windows: [0:00–0:05], [0:01–0:06], [0:02–0:07] ...
orders.groupBy((k, v) -> v.getProductId())
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))
.advanceBy(Duration.ofMinutes(1)))
.count();Session windows for user activity
Session windows group events by inactivity gap rather than clock time. Each session extends as long as activity continues within the gap period.
StreamsBuilder builder = new StreamsBuilder();
KStream<String, PageView> pageViews =
builder.stream("page-views"); // key = userId
// SESSION window — group page views; end session after 30min inactivity
// User A: views at 10:00, 10:05, 10:25 → one session (gaps < 30min)
// User A: views at 11:30 → new session (gap of 65min > 30min)
KTable<Windowed<String>, Long> sessions = pageViews
.groupByKey()
.windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
.count(Materialized.as("user-sessions"));
sessions.toStream()
.foreach((windowedKey, count) -> {
String userId = windowedKey.key();
long start = windowedKey.window().start();
long end = windowedKey.window().end();
long durationMin = (end - start) / 60_000;
log.info("User {} session: {} views, {}min duration",
userId, count, durationMin);
});Late events and grace periods
Events arriving late (after their window has passed) are handled by the grace period. After grace, late events are dropped. Monitor using the records-late-arrival metric.
// Without grace period: late events are silently dropped
TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1))
// With grace period: accept events up to 5 min after window ends
TimeWindows.ofSizeAndGrace(Duration.ofHours(1), Duration.ofMinutes(5))
// Sliding window (Kafka Streams 2.7+): every event defines a window
// around it of +/- timeDifference — better for time-range joins
SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(10))
// Check stream time (how far the topology has advanced)
// records-late-arrival metric: count of events dropped as too-late
// Via JMX: kafka.streams:type=stream-task-metrics,task-id=*
// records-late-arrival
// Test windowed streams with TopologyTestDriver
TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props);
TestInputTopic<String, OrderEvent> inputTopic =
driver.createInputTopic("order-events",
new StringSerializer(), new OrderEventSerializer());
// Inject an event 2 hours late — will be dropped by tumbling window without grace
inputTopic.pipeInput("product-1", orderEvent,
Instant.now().minus(Duration.ofHours(2)));Key Points to Remember
- 1Tumbling: non-overlapping fixed buckets. Hopping: overlapping windows sliding forward. Session: inactivity-gap based.
- 2Grace period allows late-arriving events to update already-closed windows — critical for out-of-order data.
- 3Kafka Streams uses event time (record timestamp) by default — tolerates out-of-order delivery correctly.
- 4After the grace period expires, late events are silently dropped — monitor records-late-arrival metric.
- 5Session windows merge when two sessions are closer than the inactivity gap — state store merges must be idempotent.
- 6Use TopologyTestDriver with explicit timestamps to unit-test windowing logic without real Kafka.
Interview Questions
Sign in to ask AriaWhat is the difference between tumbling, hopping, and session windows?
What happens to a late-arriving event in a Kafka Streams window without a grace period?
How does Kafka Streams advance stream time and why does it matter for windowing?
When would you use a session window instead of a tumbling window?
How would you unit test a tumbling window aggregation with specific event timestamps?
Ask Aria about Windowing in Kafka Streams
Your personal AI tutor — ask anything about this concept
Revision Status
Personal Notes
Sign in to save personal notes for this topic.
Discussion
Sign in to join the discussion.