StoatFlow

Windowing

Windowed aggregations in StoatFlow — tumbling, hopping, sliding, and session windows; emit strategies, suppression, Windowed keys, and WindowedSerdes.

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

KindFactoryOverlapResult key window
TumblingTimeWindows.ofSizeWithNoGrace(size)None — each record in exactly one windowTimeWindow
HoppingTimeWindows.ofSizeWithNoGrace(size).advanceBy(advance)Yes — advance < size means a record lands in multiple windowsTimeWindow
SlidingSlidingWindows.ofTimeDifferenceWithNoGrace(diff)Event-driven; two events share a window if |t1 - t2| <= diffTimeWindow
SessionSessionWindows.ofInactivityGapWithNoGrace(gap)Dynamic — sessions grow and merge across activitySessionWindow

Every factory has a …AndGrace(…, grace) variant that accepts late records for grace after the window's end before the window closes. The factories are static methods, so they call identically from Kotlin and Java.

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()

.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()

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)
        )

SessionWindows.ofInactivityGapAndGrace(gap, grace) adds a grace period on top of the inactivity gap before a session is considered closed.

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",
    )
}

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().

A common pattern is to fold the window bounds into the value before re-keying back to the plain key. The e-commerce example does exactly this — its mapValues reads wk.getWindow().startTime() to stamp the aggregate with its date, then toStream((wk, v) -> wk.getKey(), …) re-keys by the original customer id.

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()

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().

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:

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()))

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:

BuilderBehaviour
BufferConfig.unbounded()No size limit (a StrictBufferConfig). Use with care — memory grows with buffered records.
BufferConfig.maxRecords(n)Bound the buffer to n records.
.emitEarlyWhenFull()On overflow, emit the oldest records to make room (no data loss, but non-final values may be emitted).
.shutDownWhenFull()On overflow, shut the application down rather than emit early or lose data.
.withLoggingDisabled()Don't back the buffer with a changelog. Faster, but suppressed records are lost on restart.
BufferConfig.maxBytes(...) from Kafka Streams is not supported — StoatFlow buffers in-memory objects, not serialized bytes, so a byte limit can't be enforced accurately. Use maxRecords(...) instead; calling maxBytes(...) throws UnsupportedOperationException.

By default suppress buffers are changelog-backed for fault tolerance. suppress(...) with logging enabled requires the upstream KTable to carry serdes (provide them via Materialized on the aggregation, or disable logging with BufferConfig.withLoggingDisabled()).

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()),
    )

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.

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

Aggregations

Group records by key and fold them into running totals — count, reduce, aggregate on grouped streams and grouped tables, with the result backed by a state store.

Joins

Combine two streams or tables on a shared key — stream-stream windowed joins, stream-table enrichment lookups, table-table joins, foreign-key joins, and co-grouping.