StoatFlow

Event time and watermarks

How StoatFlow models time — event time vs processing time, record timestamps, watermark strategies, window closing, late-record handling, and event/processing-time timers.

Stream processing has to answer two different questions about time: when did this event happen and when did the runtime see it. This page explains how StoatFlow separates the two, how watermarks track event-time progress, and how that drives window closing, late-record handling, and timers. It describes the conceptual model and the public API; the mechanism that tracks and merges watermarks internally stays in the source — see Architecture for the boundary.

Event time vs processing time

Processing time is wall-clock time on the machine running the topology — when the runtime got around to this record. It's trivially available and monotonic, but it's not reproducible: re-run the same input and the processing-time boundaries land differently, because they depend on broker latency, network buffering, and how busy the runtime was. Two records produced milliseconds apart can be processed seconds apart if a batch sat in a buffer.

Event time is when the event actually happened, carried on the record itself. It's stable across replays — the same input always produces the same windowed results, regardless of how fast or slow the runtime consumed it. That reproducibility is why every stateful operator that cares about time — windowed aggregations, session windows, time-bounded joins — reasons in event time.

The runtime needs both. Processing time drives wall-clock timers and "do this every N seconds" punctuation. Event time drives windows and late-record decisions.

The record timestamp and custom extraction

Every input record carries a timestamp. By default that's the Kafka record timestamp — set by the producer when it sent the record, or by the broker on append, depending on the topic's message.timestamp.type. For many topologies that default is exactly right and you configure nothing.

When the meaningful event time lives inside the value — an eventTime field on your domain object, a timestamp parsed from a log line — supply a custom assigner. In StoatFlow this is done on the watermark strategy (the strategy combines timestamp extraction and watermark generation into one object, the same way Flink's WatermarkStrategy does). The KS-style TimestampExtractor functional interface is also supported and adapts onto a strategy internally.

The assigner receives the key, the value, and the Kafka record timestamp, and returns the event time in epoch milliseconds:

import io.stoatflow.core.topology.Consumed
import io.stoatflow.core.watermark.WatermarkStrategy
import java.time.Duration

// Event time comes from a field on the value; allow 30s of out-of-orderness.
val strategy = WatermarkStrategy
    .forBoundedOutOfOrderness<String, Order>(Duration.ofSeconds(30))
    .withTimestampAssigner { _, value, _ -> value?.eventTime ?: 0L }

builder.stream(
    "orders",
    Consumed.with(Serdes.String(), orderSerde)
        .withWatermarkStrategy(strategy),
)
The assigner may be called with a null value (tombstones) and a null key (records with no key). Guard for it — return a sensible fallback rather than dereferencing. The Java withTimestampAssigner overload shown above takes a (key, value)BiFunction; a single-argument Function<V, Long> overload is also available when only the value matters (used in the e-commerce example).

Watermarks: the "no earlier than T" claim

A watermark is a claim the runtime makes about event-time progress: "I do not expect any further records with an event time earlier than T." It's not a record and not user data — it's a moving low-water mark that tells time-sensitive operators when it's safe to act.

Watermarks are derived from the event timestamps the runtime observes. As records flow in with increasing event times, the watermark advances behind them. How far behind depends on the strategy (next section). Crucially, a watermark is a claim, not a guarantee: a record older than the current watermark can still physically arrive — that's a late record, and whether it still counts toward its window depends on that window's grace period (see below).

Watermarks are tracked per source partition — each partition has its own view of how far its event time has progressed. The runtime combines those per-partition views into a single global watermark for the whole application. Because a StoatFlow app is exactly one instance (see Architecture), there is no distributed watermark-coordination protocol to configure — the global watermark is computed in-process. The global watermark advances together with the commit barrier, so windowed results are committed alongside the watermark progress that produced them; recovery sees a consistent snapshot of "what the app has seen up to."

The current per-partition watermark state is observable at runtime on the /watermarks admin endpoint.

Idleness

If one partition stops receiving records — a quiet partition, an off-hours source — its event time stops advancing, which would hold the global watermark back and stall windows that depend on the busier partitions. A strategy configured .withIdleness(timeout) marks a partition idle after it goes quiet for that long, excluding it from the global watermark so the rest of the application keeps making event-time progress.

val strategy = WatermarkStrategy
    .forBoundedOutOfOrderness<String, Order>(Duration.ofSeconds(30))
    .withTimestampAssigner { _, value, _ -> value?.eventTime ?: 0L }
    .withIdleness(Duration.ofMinutes(2))

The idleness timeout must be positive — a zero or negative duration is rejected.

The watermark strategies

StoatFlow ships three watermark strategies, all created from WatermarkStrategy factory methods and attached per source topic via Consumed.withWatermarkStrategy(...). They're modelled on Flink's strategies.

forBoundedOutOfOrderness(maxOutOfOrderness)

The common case. Events may arrive out of order by up to maxOutOfOrderness before being considered late. The watermark is maxObservedTimestamp - maxOutOfOrderness — the larger the tolerance you set, the longer windows stay open for stragglers, at the cost of holding results back longer. maxOutOfOrderness must not be negative.

Consumed.with(Serdes.String(), orderSerde)
    .withWatermarkStrategy(
        WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(30)),
    )

forMonotonousTimestamps()

Use when events are known to arrive in strict timestamp order — a single-partition topic, log-append-time topics where the broker assigns timestamps, or events pre-sorted before production. It's equivalent to forBoundedOutOfOrderness(Duration.ZERO) with slightly less overhead. Any record whose timestamp goes backwards is treated as late.

Consumed.with(Serdes.String(), eventSerde)
    .withWatermarkStrategy(WatermarkStrategy.forMonotonousTimestamps())

noWatermarks()

Produces no watermarks at all — the watermark stays at Long.MIN_VALUE forever. Use for processing-time-only topologies, or sources without meaningful event timestamps. The trade-off is explicit: with no watermarks, event-time windows never close and event-time timers never fire. Processing-time semantics still work.

Consumed.with(Serdes.String(), eventSerde)
    .withWatermarkStrategy(WatermarkStrategy.noWatermarks())

All three accept .withTimestampAssigner(...) and .withIdleness(...), plus Flink-style .withWatermarkAlignment(group, maxDrift) for multi-source topologies where one source's event time must not run too far ahead of the others in the same alignment group. When you set neither a strategy nor an assigner, the source uses the Kafka record timestamp as event time.

Window closing

A windowed operator — a tumbling/hopping TimeWindows aggregation, a session window, a sliding window — groups records into windows by their event time. The runtime can't emit a window's final result until it's confident no more in-time records will land in it. That confidence comes from the watermark.

When the global watermark passes a window's end, the runtime knows the window can close: any record that would belong inside it would have an event time the watermark has already moved past. At that point the window's result is finalised. Until then, the window stays open and accumulates.

This is why watermark choice has a direct, visible effect on latency: a 30-second maxOutOfOrderness means a window's result is held back at least 30 seconds past its end before it can close. Tighter tolerance closes windows sooner but treats more stragglers as late.

Window specifications can carry a grace period of their own (for example TimeWindows.ofSizeAndGrace(size, grace)), which keeps a window open for late records for an extra interval after its end before it's permanently closed — see the next section.

Late records and grace

A late record is one whose event time is older than the current watermark — it arrived after the runtime had already claimed not to expect anything that old. How a windowed operator responds is governed by the grace period on the window the record belongs to:

  • Within grace — if the late record's window is still inside its configured grace window, the record is folded into the aggregate and an updated result is emitted. Grace is set on the window specification (for example TimeWindows.ofSizeAndGrace(...)), and bounds how long after a window's end late records are still accepted.
  • Past grace — once a window's end plus its grace has elapsed, the window is closed and finalised; any later record that would have belonged to it is dropped before it can change the result.

Grace is the knob that trades completeness against finality: a longer grace catches more stragglers and corrects more aggregates, but holds the window's state longer and delays the point at which the result is truly final. A window with no grace closes as soon as the watermark passes its end, dropping every straggler.

A past-grace drop is not separately metered: a record discarded because its window has already closed produces no dedicated metric or callback. The runtime does count late arrivals — records whose event time is already behind the watermark — via a debug-level counter, but that measures arrivals, not grace-exceeded drops: many counted arrivals are still folded into open windows within grace, and a window that closes with no further records produces no drop signal at all. If knowing how often stragglers are discarded matters to you, account for it in your own logic rather than relying on a built-in drop metric.

Event-time and processing-time timers

The DSL's windowed operators cover the common time-driven patterns declaratively. For logic that has to fire on a schedule independent of incoming records — flush a buffer at a deadline, expire an entry, emit a heartbeat — the Processor API lets a custom Processor register timers:

  • Event-time timers fire when the watermark advances past a registered moment. They're driven by event-time progress, so they respect the same out-of-order tolerance as windows — and, like windows, they never fire under noWatermarks().
  • Processing-time timers fire when wall-clock time passes a registered moment, regardless of whether any records are flowing.

Both deliver a callback with full state-store access in the same epoch-aligned context as record processing, so anything a timer writes commits atomically with the surrounding work (see Exactly-once semantics). The registration API and callback signatures are covered in Processor API.

For a topology-level periodic source — generating records on an interval or a cron schedule without consuming from Kafka — see Scheduled sources, which can be driven by either STREAM_TIME (event time) or WALL_CLOCK_TIME (processing time).

Examples

The e-commerce daily-customer-behaviour example (under the project's examples/ecommerce-daily-customer-behaviour module) wires custom event-time extraction into a windowed aggregation. Each source attaches a strategy that pulls event time from the record's own timestamp field, with a daily window that carries a grace period for late arrivals:

// Custom timestamp extraction off the value (single-arg Function overload)
WatermarkStrategy<String, WebActivityEvent> webActivityWatermark = WatermarkStrategy
    .<String, WebActivityEvent>forBoundedOutOfOrderness(Duration.ZERO)
    .withTimestampAssigner(v -> toUtcFromLocal(v.timestamp(), LOCAL_ZONE_ID).toEpochMilli());

KStream<String, WebActivityEvent> webActivity = builder.stream(
    WEB_ACTIVITY_SOURCE_TOPIC,
    Consumed.with(stringSerde, webActivityEventSerde)
        .withWatermarkStrategy(webActivityWatermark)
        .withName("web-activity-source"));

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

The full source includes the windowed aggregation and a downstream join enriching each daily summary against a customer-profile KTable.

Where to go next

  • Architecture — where watermark tracking sits in the engine, and the single-instance reason there's no distributed coordination
  • Windowing — tumbling, hopping, session, and sliding windows; grace and emit strategies
  • Aggregations — count, reduce, aggregate over grouped and windowed streams
  • Processor API — registering event-time and processing-time timers
  • Scheduled sources — topology-level periodic sources on stream time or wall-clock time
  • Exactly-once semantics — how windowed results and watermark progress commit atomically

State and thread-safety

How StoatFlow's global state model works — why concurrent access across lanes is safe, when you need the key-lock utility, and which store types are available.

The configuration model

How StoatFlow layers configuration — application.yaml, overlay files, environment variables, and programmatic overrides — the precedence order between them, and why some behaviour is adaptive rather than configured.