Understanding Session Windows in Apache Flink

Building on the previous article about Flink's window mechanism, this piece dives into the special case of session windows, which group user events into periods of activity separated by inactivity gaps. When the time between two events is less than a configurable sessionGap , they belong to the same window; otherwise, they start a new window.

Before Flink 1.1.0, developers had to implement session windows manually using a custom WindowAssigner and Trigger . An example in the 1.0 release (the SessionWindowing example) assigned all elements to a single global window and used a trigger that registered a timeout timer for each element, cancelling the previous timer. This approach could not handle out‑of‑order event‑time data.

Starting with Flink 1.1.0, session windows are natively supported via the SessionWindows.withGap() API, which automatically merges windows and correctly processes out‑of‑order events, borrowing ideas from Google Dataflow.

Defining a session window

For a typical e‑commerce scenario (e.g., a user browsing Taobao on a mobile device), events such as clicks, searches, and purchases are streamed to the server. These events naturally form sessions separated by gaps. Using Flink, the window can be defined as:

DataStream<T> input = …;
DataStream<R> result = input
    .keyBy(
)
    .window(SessionWindows.withGap(Time.seconds(<seconds>)))
    .apply(
); // or reduce() / fold()

Flink then assigns each element to a window based on its timestamp; if the gap between timestamps is smaller than the configured gap, the element stays in the same session, otherwise a new session window is created.

Underlying implementation

Session windows require a merging-capable window assigner. Flink introduces MergingWindowAssigner , which adds a mergeWindows method to decide which windows should be merged. The assigner creates a provisional window for each element ( [timestamp, timestamp+sessionGap) ) and later merges overlapping windows.

When windows overlap, the WindowAssigner invokes MergingWindowSet to manage the merge. The merge process combines the underlying state of the windows and updates the trigger, ensuring correct handling of event‑time watermarks and out‑of‑order data.

The core processing logic resides in WindowOperator.processElement . For each incoming element, Flink:

Assigns windows via the assigner.

If the assigner is a MergingWindowAssigner , adds the window to a MergingWindowSet , which may merge it with existing windows.

Updates the state of the resulting (merged) window using getPartitionedState and adds the element.

Invokes the trigger to decide whether to fire or purge the window.

Processes the trigger result accordingly.

The merging logic also updates timers: when windows merge, old timers are cleared and new timers are registered based on the merged window’s end time.

An optimization note: instead of creating a new window for every element and then merging, Flink could directly extend an existing window when the new element’s interval overlaps, reducing overhead.

Source code highlights

The article references JIRA FLINK‑3174 and the related pull request for the actual implementation. Key classes include:

MergingWindowAssigner – abstract class adding mergeWindows .

MergingWindowSet – tracks window‑to‑state mappings and performs merges.

WindowOperator.processElement – contains the element‑processing flow shown above.

Triggers also gain an onMerge method to adjust timers after a merge.

Conclusion

The article demonstrates how Flink’s session window support builds on window merging, enabling robust session‑based analytics for streaming data and drawing inspiration from Google Dataflow. Readers are encouraged to review the Dataflow paper for deeper theoretical background.