Windowed operations in Kafka - Step-by-Step Execution

Process Flow - Windowed operations

Stream of events

↓

Assign event timestamps

↓

Define window size & type

↓

Group events into windows

↓

Aggregate or process events per window

↓

Emit windowed results

↓

Slide or advance window

↩Back to Group events into windows

Events flow in with timestamps, get grouped into time windows, processed, and results are emitted per window.

Execution Sample

Kafka

KStream<String, Integer> stream = builder.stream("input-topic");
KTable<Windowed<String>, Long> windowedCounts = stream
  .groupByKey()
  .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(10)))
  .count();

This code counts events per key in 10-second tumbling windows.

Process Table

Step	Event Timestamp	Window Assigned	Key	Action	Windowed Count
1	00:00:03	[00:00:00 - 00:00:10)	A	Add event to window	A:1
2	00:00:07	[00:00:00 - 00:00:10)	A	Add event to window	A:2
3	00:00:12	[00:00:10 - 00:00:20)	A	Add event to new window	A:1
4	00:00:15	[00:00:10 - 00:00:20)	B	Add event to window	B:1
5	00:00:21	[00:00:20 - 00:00:30)	A	Add event to new window	A:1
6	00:00:25	[00:00:20 - 00:00:30)	A	Add event to window	A:2
7	Window [00:00:00 - 00:00:10) closes	-	-	Emit count for window	A:2
8	Window [00:00:10 - 00:00:20) closes	-	-	Emit count for window	A:1, B:1
9	Window [00:00:20 - 00:00:30) closes	-	-	Emit count for window	A:2

💡 All windows processed and counts emitted after their time range ends.

Status Tracker

Variable	Start	After 1	After 2	After 3	After 4	After 5	After 6	Final
Window [00:00:00 - 00:00:10) count for A	0	1	2	2	2	2	2	2
Window [00:00:10 - 00:00:20) count for A	0	0	0	1	1	1	1	1
Window [00:00:10 - 00:00:20) count for B	0	0	0	0	1	1	1	1
Window [00:00:20 - 00:00:30) count for A	0	0	0	0	0	1	2	2

Key Moments - 3 Insights

Why does the event at 00:00:12 start a new window instead of joining the first?

Because the first window covers [00:00:00 - 00:00:10), and 00:00:12 is outside that range, so it belongs to the next window [00:00:10 - 00:00:20). See execution_table rows 3 and 1.

Why are counts emitted only after the window closes?

Windowed operations emit results after the window time range ends to ensure all events in that window are counted. This is shown in execution_table rows 7-9.

What happens if events arrive late for a closed window?

In this example, no grace period is set, so late events are ignored or dropped. This is why windows close and emit counts strictly after their time range.

Visual Quiz - 3 Questions

Look at the execution_table at step 4, what is the window assigned to the event with key 'B'?

A [00:00:00 - 00:00:10)

B [00:00:20 - 00:00:30)

C [00:00:10 - 00:00:20)

D No window assigned yet

💡 Hint

Check the 'Window Assigned' column at step 4 in the execution_table.

At which step does the count for key 'A' in the first window reach 2?

A Step 1

B Step 2

C Step 3

D Step 6

💡 Hint

Look at the 'Windowed Count' column for key 'A' in execution_table rows 1 and 2.

If the window size was changed to 5 seconds, how would the window assigned at step 3 change?

A It would be [00:00:10 - 00:00:15)

B It would remain [00:00:10 - 00:00:20)

C It would be [00:00:05 - 00:00:10)

D No window would be assigned

💡 Hint

Window size defines the time range; smaller window means shorter intervals as seen in concept_flow.

Concept Snapshot

Windowed operations group streaming events by time intervals.
Define window size (e.g., 10 seconds) and type (tumbling, hopping).
Events are assigned to windows based on timestamps.
Aggregations (like count) happen per window.
Results emit after window closes.
Late events may be dropped or handled with grace period.

Full Transcript

Windowed operations in Kafka Streams process events grouped by time windows. Events flow in with timestamps and are assigned to windows based on these timestamps. For example, a 10-second tumbling window groups events from 0 to 10 seconds, then 10 to 20 seconds, and so on. Each event is added to its window's aggregation, such as counting occurrences per key. Counts are emitted only after the window time ends to ensure completeness. If events arrive late, they may be ignored if no grace period is set. This process repeats as new events arrive and windows slide forward.

Practice

1. What is the main purpose of windowed operations in Kafka Streams?

easy

A. To compress data for storage

B. To group data into time intervals for easier analysis

C. To encrypt data streams

D. To sort data alphabetically

Solution

1. Step 1: Understand windowed operations

   Windowed operations group data based on time intervals, making it easier to analyze events occurring within those intervals.

2. Step 2: Compare options

   Sorting, encrypting, and compressing are unrelated to windowing, which focuses on time-based grouping.

3. Final Answer:

   To group data into time intervals for easier analysis -> Option B

4. Quick Check:

   Windowed operations = time grouping [OK]

Hint: Windowed means grouping by time intervals [OK]

Common Mistakes:

• Confusing windowing with sorting
• Thinking windowing encrypts data
• Assuming windowing compresses data

2. Which of the following is the correct syntax to create a tumbling window of 5 minutes in Kafka Streams DSL?

easy

A. SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))

B. SessionWindows.of(Duration.ofMinutes(5))

C. TimeWindows.of(Duration.ofMinutes(5))

D. Windowed.of(Duration.ofMinutes(5))

Solution

1. Step 1: Identify tumbling window syntax

   Tumbling windows in Kafka Streams are created using TimeWindows.of() with a duration.

2. Step 2: Check options

   SessionWindows is for session windows, SlidingWindows is for sliding windows, and Windowed is not a window type class.

3. Final Answer:

   TimeWindows.of(Duration.ofMinutes(5)) -> Option C

4. Quick Check:

   Tumbling window = TimeWindows.of() [OK]

Hint: Tumbling windows use TimeWindows.of() syntax [OK]

Common Mistakes:

• Using SessionWindows for tumbling windows
• Confusing SlidingWindows with tumbling windows
• Using non-existent Windowed class

3. Given the following Kafka Streams code snippet, what will be the output count for the window ending at timestamp 60000?

stream.groupByKey()
      .windowedBy(TimeWindows.of(Duration.ofSeconds(30)))
      .count()
      .toStream()
      .foreach((windowedKey, count) -> System.out.println(windowedKey.window().end() + ": " + count));

Assume events with keys arrive at timestamps 10000, 20000, 35000, 40000, and 55000.

medium

A. 3

B. 4

C. 2

D. 5

Solution

1. Step 1: Identify window intervals

   Windows are 30 seconds long: [0-30000), [30000-60000), etc. Events at 10000 and 20000 fall in first window; 35000, 40000, 55000 fall in second window ending at 60000.

2. Step 2: Count events in window ending at 60000

   Events at 35000, 40000, and 55000 are inside the window ending at 60000, so count is 3.

3. Final Answer:

   3 -> Option A

4. Quick Check:

   Events in 30s window = 3 [OK]

Hint: Count events in the correct time window interval [OK]

Common Mistakes:

• Counting events from previous window
• Misunderstanding window boundaries
• Including events outside the window

4. Identify the error in this Kafka Streams windowed count code snippet:

stream.groupByKey()
      .windowedBy(SessionWindows.of(Duration.ofMinutes(1)))
      .count()
      .toStream()
      .foreach((windowedKey, count) -> System.out.println(windowedKey.window().start() + ": " + count));

medium

A. windowedBy() cannot be used with groupByKey()

B. toStream() is not available after count()

C. count() must be called before windowedBy()

D. SessionWindows.of() requires a gap duration, not a window size

Solution

1. Step 1: Check SessionWindows.of() usage

   SessionWindows require a gap duration (inactivity gap), not a fixed window size in of().

2. Step 2: Validate other method calls

   groupByKey() works with windowedBy(), count() is correctly placed, and toStream() is valid after count().

3. Final Answer:

   SessionWindows.of() requires a gap duration, not a window size -> Option D

4. Quick Check:

   SessionWindows need gap duration [OK]

Hint: SessionWindows need gap duration, not fixed size [OK]

Common Mistakes:

• Using fixed duration in SessionWindows.of()
• Misplacing count() before windowedBy()
• Assuming toStream() is invalid after count()

5. You want to count user clicks in overlapping 1-minute windows that advance every 30 seconds using Kafka Streams. Which window type and configuration should you use?

hard

A. Use TimeWindows.of(Duration.ofMinutes(1)).advanceBy(Duration.ofSeconds(30))

B. Use SlidingWindows.ofTimeDifferenceAndGrace(Duration.ofMinutes(1), Duration.ofSeconds(0)).advanceBy(Duration.ofSeconds(30))

C. Use SessionWindows.withGap(Duration.ofSeconds(30))

D. Use TimeWindows.of(Duration.ofSeconds(30)).advanceBy(Duration.ofMinutes(1))

Solution

1. Step 1: Identify overlapping window type

   Overlapping windows that slide are created with TimeWindows.of() and advanceBy() to set the slide interval.

2. Step 2: Match configuration to requirement

   Window size is 1 minute, sliding every 30 seconds, so TimeWindows.of(1 minute).advanceBy(30 seconds) fits perfectly.

3. Final Answer:

   Use TimeWindows.of(Duration.ofMinutes(1)).advanceBy(Duration.ofSeconds(30)) -> Option A

4. Quick Check:

   Sliding windows = TimeWindows with advanceBy [OK]

Hint: Sliding windows use TimeWindows with advanceBy() [OK]

Common Mistakes:

• Using SlidingWindows class which is deprecated
• Confusing session windows with sliding windows
• Swapping window size and advanceBy durations