Windowing

Windowing groups records by key and by time, so an aggregation produces one result per key per window instead of a single running total. You reach a windowed aggregation by calling .windowedBy(...) on a KGroupedStream — the same grouped type you get from groupByKey() / groupBy(). The result is a KTable whose key is a Windowed<K>: the original key plus the window bounds.

Windows are reasoned about in event time, not wall-clock time — when a window closes is driven by the watermark, not by the clock on the machine. Read Event time and watermarks for the model that underpins everything on this page.

Window kinds at a glance

Every factory has a …AndGrace(…, grace) variant that accepts late records for grace after the window's end before the window closes. A window closes when the watermark passes its end plus the grace period — for a session, its last event plus the inactivity gap plus grace. A record that arrives for a closed window is dropped, not added; it is counted in stoatflow.dropped.records.total with reason=late_window (or late_session). The watermark trails the newest event time by the source's maximum out-of-orderness (10 s by default) and is kept across restarts, so a window closed before a restart stays closed after it. The factories are static methods, so they call identically from Kotlin and Java.

The same six records, under all four kinds — trace any record straight down to compare:

Tumbling and hopping windows

TimeWindows covers both fixed-size cases. A tumbling window has no overlap — its advance equals its size. Calling .advanceBy(...) with an interval smaller than the size turns it into a hopping window, where each record falls into several overlapping windows.

import io.stoatflow.core.topology.Grouped
import io.stoatflow.core.topology.TimeWindows
import io.stoatflow.core.topology.Windowed
import java.time.Duration

// Tumbling: count page views per user in fixed 5-minute windows
val tumbling: KTable<Windowed<String>, Long> =
    pageViews
        .groupByKey(Grouped.with(Serdes.String(), pageViewSerde))
        .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
        .count()

// Hopping: 5-minute windows that advance every 1 minute (each view lands in 5 windows)
val hopping: KTable<Windowed<String>, Long> =
    pageViews
        .groupByKey(Grouped.with(Serdes.String(), pageViewSerde))
        .windowedBy(
            TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))
                .advanceBy(Duration.ofMinutes(1)),
        )
        .count()

import io.stoatflow.core.topology.Grouped;
import io.stoatflow.core.topology.TimeWindows;
import io.stoatflow.core.topology.Windowed;
import java.time.Duration;

// Tumbling: count page views per user in fixed 5-minute windows
KTable<Windowed<String>, Long> tumbling =
    pageViews
        .groupByKey(Grouped.with(Serdes.String(), pageViewSerde))
        .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
        .count();

// Hopping: 5-minute windows that advance every 1 minute (each view lands in 5 windows)
KTable<Windowed<String>, Long> hopping =
    pageViews
        .groupByKey(Grouped.with(Serdes.String(), pageViewSerde))
        .windowedBy(
            TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))
                .advanceBy(Duration.ofMinutes(1)))
        .count();

.advanceBy(...) must be positive and no larger than the window size; an advance equal to the size is just a tumbling window again. To accept late records, swap the factory for TimeWindows.ofSizeAndGrace(size, grace) — the window then stays open for grace past its end before closing.

count(), reduce(...), and aggregate(...) are all available on the windowed stream, with the same Kotlin-lambda / Java-functional-interface overloads as the non-windowed aggregations (see Aggregations). count() returns KTable<Windowed<K>, Long>; reduce and aggregate return your value type.

Sliding windows

A sliding window is defined by a time difference rather than a fixed grid position. Two events belong to the same window when their event-time difference is within the configured difference; windows are positioned relative to each event. This is the KIP-450 event-driven model — roughly two to three windows per event, not one window per millisecond of advance.

import io.stoatflow.core.topology.SlidingWindows
import java.time.Duration

val slidingCounts: KTable<Windowed<String>, Long> =
    events
        .groupByKey(Grouped.with(Serdes.String(), eventSerde))
        .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(10)))
        .count()

import io.stoatflow.core.topology.SlidingWindows;
import java.time.Duration;

KTable<Windowed<String>, Long> slidingCounts =
    events
        .groupByKey(Grouped.with(Serdes.String(), eventSerde))
        .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(10)))
        .count();

For a 10-minute difference, events at t=0 and t=5min share a window (5 ≤ 10), while events at t=0 and t=15min do not (15 > 10). Use SlidingWindows.ofTimeDifferenceAndGrace(diff, grace) to admit late records.

Session windows

Session windows are dynamic. A session grows as long as records keep arriving within the inactivity gap; once activity pauses longer than the gap, the session closes and the next record starts a fresh one. When a record bridges two existing sessions, they merge.

Because sessions merge, aggregate(...) on a session window requires a session merger — a function that combines the aggregates of two sessions being joined. count() and reduce(...) supply this implicitly (counts sum; the reducer doubles as the merger), but aggregate makes you pass it explicitly.

import io.stoatflow.core.topology.SessionWindows
import java.time.Duration

// Count user actions per session (30-minute inactivity gap)
val sessionCounts: KTable<Windowed<String>, Long> =
    userActions
        .groupByKey(Grouped.with(Serdes.String(), actionSerde))
        .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
        .count()

// Aggregate with an explicit session merger
val sessionActions: KTable<Windowed<String>, List<Action>> =
    userActions
        .groupByKey(Grouped.with(Serdes.String(), actionSerde))
        .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
        .aggregate(
            { emptyList() },                                   // initializer
            { _, value, agg -> agg + value },                 // aggregator
            { _, agg1, agg2 -> agg1 + agg2 },                 // session merger (required)
        )

import io.stoatflow.core.topology.SessionWindows;
import java.util.ArrayList;
import java.util.List;
import java.time.Duration;

// Count user actions per session (30-minute inactivity gap)
KTable<Windowed<String>, Long> sessionCounts =
    userActions
        .groupByKey(Grouped.with(Serdes.String(), actionSerde))
        .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
        .count();

// Aggregate with an explicit session merger
KTable<Windowed<String>, List<Action>> sessionActions =
    userActions
        .groupByKey(Grouped.with(Serdes.String(), actionSerde))
        .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
        .aggregate(
            ArrayList::new,                                   // initializer
            (key, value, agg) -> { agg.add(value); return agg; },  // aggregator
            (key, agg1, agg2) -> { agg1.addAll(agg2); return agg1; });  // session merger (required)

SessionWindows.ofInactivityGapAndGrace(gap, grace) adds a grace period on top of the inactivity gap before a session is considered closed. A closed session is final: a later record that falls within its gap never re-opens it, but starts a new session.

Windowed keys

Every windowed aggregation keys its output by Windowed<K> — the original key plus a Window (a TimeWindow for time/sliding aggregations, a SessionWindow for sessions). The window is a half-open interval [start, end): it includes start and excludes end.

windowedCounts.toStream().forEach { windowedKey, count ->
    println(
        "key=${windowedKey.key} " +
            "window=${windowedKey.windowStartTime()}..${windowedKey.windowEndTime()} " +
            "count=$count",
    )
}

windowedCounts.toStream().forEach((windowedKey, count) ->
    System.out.println(
        "key=" + windowedKey.getKey()
            + " window=" + windowedKey.windowStartTime() + ".." + windowedKey.windowEndTime()
            + " count=" + count));

Windowed<K> exposes windowStart() / windowEnd() (epoch millis) and windowStartTime() / windowEndTime() (Instant). In Kotlin the underlying key and window are properties (windowedKey.key, windowedKey.window); from Java they are the generated accessors getKey() and getWindow().

When results are emitted

By default a windowed aggregation emits a result on every update — each record that changes a window's aggregate produces a new downstream record (continuous refinement). For high-volume windows that is a lot of intermediate output. EmitStrategy lets you switch to emitting only the final result when the window closes.

import io.stoatflow.core.topology.EmitStrategy
import io.stoatflow.core.topology.TimeWindows
import java.time.Duration

val finalCounts: KTable<Windowed<String>, Long> =
    pageViews
        .groupByKey(Grouped.with(Serdes.String(), pageViewSerde))
        .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30)))
        .emitStrategy(EmitStrategy.onWindowClose())
        .count()

import io.stoatflow.core.topology.EmitStrategy;
import io.stoatflow.core.topology.TimeWindows;
import java.time.Duration;

KTable<Windowed<String>, Long> finalCounts =
    pageViews
        .groupByKey(Grouped.with(Serdes.String(), pageViewSerde))
        .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30)))
        .emitStrategy(EmitStrategy.onWindowClose())
        .count();

EmitStrategy.onWindowClose() produces exactly one output per window per key, at the cost of latency equal to the grace period — the window can't close until the watermark passes windowEnd + grace. It is available on the windowed stream types (TimeWindowedKStream — which covers both fixed-size and sliding windows — and SessionWindowedKStream) and requires watermark propagation to detect closure. The default is EmitStrategy.onWindowUpdate().

Kafka Streams decides lateness per aggregation, from the newest event time that reached it, so its results can differ from StoatFlow's for out-of-order input or behind a filter (KSC-132).

A result carries the event time of its window — the latest record time in it (a session: its end, for updates and merge retractions too) — so you can window it again downstream. When that happens across a re-key, the downstream window counts every upstream final and closes about two commit intervals later per step. A graceful shutdown emits every window the watermark has closed, in every step; a window still open is emitted once the watermark passes it after the next start.

A downstream window must close no earlier than the upstream window that feeds it — downstream end + grace ≥ upstream end + grace — because a final can carry a time early in its window and is judged against a watermark just below the upstream close. Re-aggregating into larger, aligned windows (1 minute → 1 hour) needs only the upstream grace. Into smaller or hopping windows, add the size difference: 1 hour → 1 minute needs the upstream grace plus 59 minutes, or a final whose latest record came early in the hour is dropped as late.

How long windows are retained

A windowed aggregation is backed by a state store, and that store keeps windows for a retention period — by default the window size plus the grace period, overridable with Materialized.withRetention(...). Windows are expired once they fall far enough behind — though, as the next paragraph explains, "far enough" is not exactly the configured retention on a persistent store: the segment holding the boundary is kept in full, so a window just past retention can still answer queries until its whole segment ages out.

Three regimes decide which term sets the interval, and a multiplier only means something alongside the condition that makes it bind:

Worked examples, all computed from the shipped formula:

Two of those are worth reading twice, because they contradict the shorthands people reach for. Below about three minutes of retention it is not always true that the 60-second floor governs — a 90-second window inside a 2-minute retention makes the window-size term the largest. And the retention + 60s regime is not a Kafka Streams quirk: StoatFlow has had the same floor since ADR-141, which is why a 30-second persistent session store keeps up to 3× what you configured rather than 4/3×.

A related shorthand to retire: being on the floor does not mean the store holds one segment. The number of live segments is set by where the watermark sits: retention / segmentInterval whole intervals plus the one holding the boundary when it lands on a segment edge, and ceil(retention / segmentInterval) + 1 in the worst phase — the same number when the retention divides evenly, one higher when it does not, since segments are fixed-width and epoch-aligned. Size against the maximum. A single segment is never guaranteed — 30 seconds of retention on a 60-second interval still spans two whenever the watermark is off a boundary. That assumes one uniform layout: after a retention or interval change, existing segments keep the span they were recorded with (ADR-141), so until they age out the live set is a mix of widths and the formula does not apply — size against retention + the widest existing span through that transition.

The mechanism behind the slack, if you prefer it to the arithmetic: a segment is dropped only once its entire time range has fallen at or before watermark − retention. The segment containing that boundary is retained in full. That is the slack.

Retention under EmitStrategy.onWindowClose()

There is a third dimension the bound above does not cover. Under the default onWindowUpdate(), expiry runs on the committed watermark, because every closed window has already been forwarded. Under onWindowClose(), expiry is gated on the durable emit frontier instead, so a window is never expired before its final result has been emitted:

Session state is the odd one out under the default strategy: it expires on observed stream time rather than the committed watermark, so it sheds earlier than a time-windowed store on the same data — and not merely slightly. The watermark is the minimum across partitions less the out-of-orderness bound, so if one partition runs ahead while another stalls, session expiry leads time-window expiry by that whole gap. Worth knowing if you are comparing the two side by side, or querying session state near the retention boundary.

The extra term is normally small and is zero-slack by design — the gate exists precisely so expiry can never overtake an un-emitted window close. But it is not bounded by the retention contract, so size disk for it if you run onWindowClose() on a large keyspace.

The gate lives in the processor, not the store, so this applies to the in-memory stores too: under onWindowClose() they no longer keep exactly retention either. The exact-1× property is an onWindowUpdate() statement.

For sizing guidance see Tuning.

Suppression

EmitStrategy is the declarative way to ask for final-only output. KTable.suppress(...) is the lower-level control: it holds updates in a buffer and you choose both the release condition and how the buffer behaves when it fills.

For windowed tables, Suppressed.untilWindowCloses(bufferConfig) buffers every update until the window closes — equivalent in effect to onWindowClose(), but with explicit buffer-overflow control. It closes with the grace of the windowed aggregation the table derives from, even after a re-key, and is refused at build for a table that derives from none:

import io.stoatflow.core.topology.Suppressed
import io.stoatflow.core.topology.Suppressed.BufferConfig
import io.stoatflow.core.topology.TimeWindows
import java.time.Duration

val suppressed: KTable<Windowed<String>, Long> =
    pageViews
        .groupByKey(Grouped.with(Serdes.String(), pageViewSerde))
        .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30)))
        .count()
        .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))

import io.stoatflow.core.topology.Suppressed;
import io.stoatflow.core.topology.Suppressed.BufferConfig;
import io.stoatflow.core.topology.TimeWindows;
import java.time.Duration;

KTable<Windowed<String>, Long> suppressed =
    pageViews
        .groupByKey(Grouped.with(Serdes.String(), pageViewSerde))
        .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30)))
        .count()
        .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()));

untilWindowCloses needs a windowed aggregation upstream of the suppress. On a table with no windowed ancestor it throws TopologyValidationException at build time ("Window close time is only defined for windowed computations"), as Kafka Streams does. The window's close time comes from the window definition (size plus grace; for a session window, gap plus grace) and is found through any operators in between, such as filter, mapValues or a toStream()/toTable() round trip. Where windowed streams are merged ahead of the suppress, the largest grace among them applies.

For non-windowed tables, Suppressed.untilTimeLimit(duration, bufferConfig) rate-limits per key: at most one update per key per interval, keeping only the most recent value within each interval.

Buffer configuration

suppress needs a BufferConfig — a type nested inside Suppressed (Suppressed.BufferConfig, matching Kafka Streams) — describing how much to hold and what to do on overflow:

By default suppress buffers are changelog-backed for fault tolerance (BufferConfig.withLoggingDisabled() turns that off). The buffer uses the upstream KTable's serdes, and falls back to the configured default.key.serde / default.value.serde when none were set upstream. That includes windowed tables, where Kafka Streams fails instead (KAFKA-9259). If the table's values don't match the default value serde, set one upstream, for example with Materialized.withValueSerde(...).

Serializing windowed keys to a sink

To write a windowed table to a topic, the Windowed<K> key needs a serde. WindowedSerdes wraps the inner key serde with the window bounds:

import io.stoatflow.core.topology.Named
import io.stoatflow.core.topology.Produced
import io.stoatflow.core.topology.Windowed
import io.stoatflow.core.topology.WindowedSerdes
import org.apache.kafka.common.serialization.Serdes

val windowedKeySerde = WindowedSerdes.timeWindowedSerdeFrom(Serdes.String())

windowedCounts
    .toStream(Named.`as`("to-stream"))
    .to(
        "windowed-counts",
        Produced.`as`<Windowed<String>, Long>("sink")
            .withKeySerde(windowedKeySerde)
            .withValueSerde(Serdes.Long()),
    )

import io.stoatflow.core.topology.Named;
import io.stoatflow.core.topology.Produced;
import io.stoatflow.core.topology.Windowed;
import io.stoatflow.core.topology.WindowedSerdes;
import org.apache.kafka.common.serialization.Serdes;

var windowedKeySerde = WindowedSerdes.timeWindowedSerdeFrom(Serdes.String());

windowedCounts
    .toStream(Named.as("to-stream"))
    .to(
        "windowed-counts",
        Produced.<Windowed<String>, Long>as("sink")
            .withKeySerde(windowedKeySerde)
            .withValueSerde(Serdes.Long()));

Use WindowedSerdes.sessionWindowedSerdeFrom(inner) for session-windowed keys so deserialization reconstructs SessionWindow instances. Both factories take a Serde<T> for the inner key (a StoatFlow deviation from Kafka Streams, which takes a Class<T>); the window size is omitted because the bounds are encoded in the serialized bytes.

The same serde is how you window an already-windowed key again: declare it on the grouping, for example groupByKey(Grouped.with(WindowedSerdes.timeWindowedSerdeFrom(Serdes.String()), valueSerde)), or on the aggregation's Materialized, for example count(Materialized.with(WindowedSerdes.timeWindowedSerdeFrom(Serdes.String()), Serdes.Long())). Without a declared inner key serde in either place, the nested aggregation is rejected at build time with TopologyValidationException. A groupByKey(Grouped) that sets no key serde — Grouped.as(name), say — keeps the key serde the stream already declared, as groupByKey() does.

merge accepts windowed streams whose key serdes differ, as Kafka Streams does. Both sides carry the same key type, so what one side declares describes the other: a side with no key serde takes the other side's, the same serde carries through, and a full WindowedSerdes.timeWindowedSerdeFrom(inner) counts as the windowed key over inner. Only when the two sides declare different serdes (or window kinds) is the merged key undeclared — then declare the windowed key serde downstream, on the sink's Produced, a Repartitioned, Grouped or Materialized. A windowed key that reaches a site with no declared serde fails the build with TopologyValidationException, naming the site and the fix.

A complete windowed aggregation

The e-commerce example aggregates per-customer activity into a daily window, then re-keys back to the customer id and joins a profile table. Its core is a tumbling daily window with a 15-minute grace period:

// Daily tumbling windows, 15-minute grace for late events
TimeWindows dailyWindows =
    TimeWindows.ofSizeAndGrace(Duration.ofDays(1), Duration.ofMinutes(15));

TimeWindowedKStream<String, WebActivityOrPurchase> windowedStream = mergedStreams
    .groupBy(
        (k, v) -> v.purchase() != null ? v.purchase().customerId() : v.webActivity().customerId(),
        Grouped.with("grouped-by-customer-id", stringSerde, webActivityOrPurchaseSerde))
    .windowedBy(dailyWindows);

KTable<Windowed<String>, DailyAggregates> dailyAggregation = windowedStream
    .aggregate(
        () -> new DailyAggregates(/* empty */),
        (key, value, agg) -> /* fold value into agg */ agg,
        Named.as("daily-aggregation"),
        Materialized.<String, DailyAggregates, StateStore>as("daily-aggregation-store")
            .withValueSerde(dailyAggregatesSerde));

The full source — including how it reads wk.getWindow().startTime() to stamp each aggregate with its date and toStream((wk, v) -> wk.getKey(), …) to re-key — is in examples/ecommerce-daily-customer-behaviour.

Where to go next

• Aggregations — count, reduce, aggregate and the grouped types that windowing builds on
• Event time and watermarks — how StoatFlow decides when a window closes
• Serdes — serializers for keys and values, including custom value types
• State stores — the window/session stores backing these aggregations
• Testing — advancing the watermark to drive window closes in tests