Event time and watermarks

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

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.
WatermarkStrategy<String, Order> strategy = WatermarkStrategy
    .<String, Order>forBoundedOutOfOrderness(Duration.ofSeconds(30))
    .withTimestampAssigner((key, value) -> value.getEventTime());

builder.stream(
    "orders",
    Consumed.with(Serdes.String(), orderSerde)
        .withWatermarkStrategy(strategy));

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

The runtime does not re-sort records within a partition by timestamp: each partition is processed in offset order, and timestamp order is chosen only between partitions (the partition whose next record is oldest goes first). So a record whose event time is older than records before it in the same partition is treated exactly like any other out-of-order record — the watermark strategy's out-of-orderness and the window's grace period decide whether it still counts. Kafka Streams and Flink behave the same way.

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

WatermarkStrategy<String, Order> strategy = WatermarkStrategy
    .<String, Order>forBoundedOutOfOrderness(Duration.ofSeconds(30))
    .withTimestampAssigner((key, value) -> value.getEventTime())
    .withIdleness(Duration.ofMinutes(2));

The idleness timeout must be positive — withIdleness rejects a zero or negative duration. It applies to the partitions of the source it is set on; sources that set none use the application-wide stoatflow.watermark.idleness-timeout-ms (5 minutes by default; 0 disables idleness for those sources — see Core configuration). The start-up warning about 0 names the sources actually left uncovered, so a topology where every source sets its own timeout is silent.

Records stamped in the future

The dispatcher serves partitions oldest-next-record first, and each partition in offset order. A record whose timestamp is far ahead of the others therefore waits at the head of its partition until every older record elsewhere has gone first, and its partition dispatches nothing meanwhile.

The partition waits; event time does not. With the built-in strategies (forBoundedOutOfOrderness, forMonotonousTimestamps), a partition whose next record is held back this way counts towards the application's watermark as if that record had been processed — the watermark the partition is committed to reach once it is. The application's watermark therefore follows the other partitions, and windows close, suppress emits and event-time timers fire as they would without the held record. It takes about a second from the record starting to wait. Once the record is processed, the partition counts normally again. The record itself keeps its place: nothing is dispatched out of order, and committed offsets and partition times still describe processed records only.

What this changes for lateness: records of that partition that arrive behind the held one are judged against the other partitions' progress. A record in order behind it is never late. One far out of order behind it (older than what the other partitions have already closed) is late — as it would be once the held record is processed, or once the partition is declared idle. This is Flink's outcome for a split that runs ahead.

A custom WatermarkGenerator gets none of this: there, the held partition's watermark stands still, and with it the application's event time, until the record is processed or the partition is declared idle by the idleness timeout above (5 minutes by default).

stoatflow.watermark.preview.partitions counts the partitions held this way — any partition whose next record waits behind older records, so it also counts one far ahead during a skewed catch-up — and /watermarks shows such a partition's previewWatermarkMs (what it counts with) and headHeldSinceMs next to its own watermarkMs. StoatFlow also logs a warning naming the partition when its next record is stamped more than a minute ahead of the wall clock: that the partition waits behind older records — and whether event time follows the other partitions or, with a custom generator, stands still — or, when no other partition is behind it (a single partition, or every partition ahead), that the timestamps look wrong. The usual cause is a producer with a wrong clock or a timestamp field in the wrong unit.

If you implement WatermarkStrategy yourself, idlenessTimeout() follows the same rule with one extra case: returning null inherits the application-wide default rather than disabling idleness, and returning Duration.ZERO is how a custom strategy switches idleness off for its own source.

If your own WatermarkStrategy extracts the event time from anything but the record timestamp — the key, the value or the headers — its hasCustomTimestampAssigner() must return true. StoatFlow trusts that flag: it decides whether partitions are ordered by their extracted event time, and it is what lets a held partition count towards the watermark by that event time. Left at false, a held partition counts with its records' Kafka timestamps, and if the extracted event times are earlier, the watermark (which never goes back) passes them and the records are dropped as late. The built-in strategies set it for you when you call withTimestampAssigner(...).

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

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

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 timers never fire. Processing-time semantics still work.

A topology that reads a source with noWatermarks() and has an operation that runs on the watermark — a windowed aggregation, suppress, or a stream-stream join — is refused at startup, naming the sources and the operations: their windows would never close, late records would never be dropped, and their state would never expire.

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

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 — see Watermark alignment below. When you set neither a strategy nor an assigner, the source uses the Kafka record timestamp as event time.

Watermark alignment

.withWatermarkAlignment(group, maxAllowedWatermarkDrift[, updateInterval]) — Flink's API, signature for signature — keeps the sources of an alignment group from drifting apart in event time. StoatFlow already dispatches the oldest buffered record first across all partitions, which keeps event time aligned in most cases. Alignment adds an explicit bound for the case ordering alone cannot cover: a sparse source, whose quiet stretches let the others run ahead.

val aligned = WatermarkStrategy
    .forBoundedOutOfOrderness<String, Order>(Duration.ofSeconds(5))
    .withIdleness(Duration.ofMinutes(1))
    .withWatermarkAlignment("orders-and-payments", Duration.ofSeconds(30))

WatermarkStrategy<String, Order> aligned = WatermarkStrategy
    .<String, Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withIdleness(Duration.ofMinutes(1))
    .withWatermarkAlignment("orders-and-payments", Duration.ofSeconds(30));

What it does. A partition whose watermark is more than maxAllowedWatermarkDrift ahead of another partition of its group — a lagging partition, one that has delivered records and is not idle — is held. A held partition keeps being fetched; alignment never pauses it. Its records are still dispatched in timestamp order, and when its next record is the oldest one buffered anywhere in the application:

• while some lagging partition has nothing buffered, the hold waits: nothing is dispatched, because that partition may still deliver a record older than the one waiting, or go idle;
• once every lagging partition has data, none of them can advance before the waiting record is processed, so it is dispatched.

A waiting hold stops the whole application, not just its group. Records are never reordered, so nothing may overtake the oldest buffered record: every source waits, in every group. Event-time timers and scheduled sources at or after that record wait with it; processing-time timers, punctuators and commits carry on.

What the drift bound means. Against a lagging partition with nothing buffered, the watermark drift stays within maxAllowedWatermarkDrift. Against one with data, the guarantee is timestamp order — the held partition never gets ahead of the lagging one's records. The two coincide when the group's sources share an out-of-orderness bound, so give them the same bound: a group whose bounds differ by more than the drift keeps the smaller-bound source held for good, and StoatFlow warns about it when it builds the topology (rule alignment-mixed-watermark-bounds).

Sizing the drift. For a sparse source that sends a record every G on average, with out-of-orderness bounds b(sparse) and b(held), a hold never has to wait in steady state when

maxAllowedWatermarkDrift ≥ G + b(sparse) − b(held)

Below that, the application stops for most of each gap — with equal bounds, G = 30 s and a 5 s drift, for about 25 s of every 30. The alternative is a shorter withIdleness(...) on the sparse source: each wait then ends after about that timeout, at a price — once the source is declared idle the watermark moves on without it, so a record it sends afterwards can be late.

Idleness is required. A lagging partition that never sends again releases a hold only by being declared idle, so every aligned source needs an idleness timeout. StoatFlow refuses to build a topology where an aligned source has none — alignment-idleness-off, an error by default; see Topology validation. While a hold waits, the idleness clocks of partitions that still have buffered records stand still — waiting behind the hold is not silence — while empty partitions keep counting, so the silent source still idles out and releases the hold. A partition with records waiting also stays in its alignment group after it is declared idle, so it remains held — waiting behind older records is not silence.

Watching it. /watermarks marks each held partition (alignmentHeld) and, while a hold waits, names it in alignmentBinding with its group, its lagging partitions that have nothing buffered, and since when. stoatflow.watermark.alignment.held.time.total counts the seconds a group kept the application waiting — its rate() is the fraction of time — and a hold that stops all processing for more than a minute is logged once with the sizing rule — as is a group whose holds keep the application waiting at least half of a five-minute window, the shape of a sparse source that sends more often than once a minute. updateInterval (1 s by default; the application uses the smallest over its aligned sources) is how often the held partitions are recomputed, so a partition that runs ahead can overshoot by up to one interval before it is held.

The TopologyTestDriver does not model alignment — it has no partitions and one watermark — but the two topology rules above run there too.

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.

The whole cycle on one timeline. The axis is event time; the circled numbers are the order the records arrived in, which is what actually drives the watermark:

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 the watermark has passed a window's end plus its grace, the window is closed and finalised; any later record that would have belonged to it is dropped before it can change the result. The watermark is the one clock for both the close and the drop, so a closed window never changes again — and it is kept across restarts. Kafka Streams decides this per aggregation, from the newest event time that reached it, so for out-of-order input or behind a filter the two can count different records into a window (KSC-132).

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 metered, and it is a different quantity from a late arrival. A record discarded because its window has already closed is counted in stoatflow.dropped.records.total with reason="late_window" (or "late_session" for session windows) — the counterpart of Kafka Streams' dropped-records — and logged at debug level; there is no callback and the record is not sent to a dead-letter topic. Separately, the runtime counts late arrivals — records whose event time is already behind the watermark — in the debug-level stoatflow.watermark.late.records.total; most of those are still folded into open windows within grace, so that counter measures how out-of-order the input is, not how much of it was lost. Alert on the first; tune out-of-orderness with the second. Both are listed in the metrics reference.

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