How do session windows work, including session merging and the inactivity gap?

## Concept A **session window** captures a burst of activity. Instead of fixed clock boundaries, a session is defined by **periods of activity separated by periods of inactivity**. This matches real user behavior: a user clicks several times, goes idle, comes back later — each burst is a session. `SessionWindows.ofInactivityGapAndGrace(Duration inactivityGap, Duration grace)`. ## The inactivity gap Two records (for the same key) belong to the same session if the time between them is `<= inactivityGap`. If the gap between a new record and the nearest existing session boundary exceeds `inactivityGap`, a **new** session starts. So a session's `[start, end]` is `[min timestamp, max timestamp]` of the records it contains, and its effective duration is variable. ## Growth When an in-gap record arrives, the session's `end` (and possibly `start`, for out-of-order earlier records) is extended to include it, and the aggregate is updated. The old session entry (with its previous boundaries) is deleted and a new one with the new boundaries is written — sessions are keyed by their boundaries, so changing boundaries changes the key. ## Merging — the defining feature With **out-of-order** data, a record can land **between** two existing sessions and be within `inactivityGap` of both. Those two sessions must then **merge** into a single session covering both. Kafka Streams: 1. Finds all sessions within `gap` of the new record. 2. Removes them from the store. 3. Creates one merged session whose boundaries span all of them and whose aggregate is the combination of all their values. This is why session aggregation requires a **merger** in addition to the aggregator: ``` .aggregate(initializer, aggregator, merger, Materialized...) ``` The `merger: (key, aggLeft, aggRight) -> aggMerged` tells Streams how to fold two sessions' partial aggregates together (e.g. sum the counts). ## Storage Sessions live in a **SessionStore** (a specialized RocksDB store, segment-based like window stores) keyed by `(key, sessionStartTime, sessionEndTime)`. Because boundaries are part of the key, merges manifest as delete-old + put-new operations, reflected in the changelog topic. ## Grace and retention `grace` defines how long after a session's end late records may still extend or merge it. Once `streamTime > sessionEnd + grace`, the session is final; later records for it are dropped. The store retention must be at least `inactivityGap + grace`. ## suppress for final-only output Like other windows, intermediate session updates stream out as a changelog. Pair with `suppress(Suppressed.untilWindowCloses(...))` to emit one final record per session after it closes. ## Edge cases - A single isolated record is a valid session of zero duration (`start == end`). - Heavy out-of-order data causes frequent merges → extra store churn and changelog traffic. - Different keys have completely independent session boundaries.