Three clocks: stream time, Flink watermarks, and the one watermark StoatFlow computes

Run the same five-minute windowed count over the same records on Kafka Streams, Flink and StoatFlow, and you get three different answers to the question "is this window done?" — not because the windows differ, but because each engine keeps a different clock. Kafka Streams' clock is the highest record timestamp a task has seen. Flink's is a watermark that every operator derives for itself as the minimum over its inputs. StoatFlow's is one watermark computed once, at the edge: the minimum across non-idle partitions, minus an out-of-orderness bound, delivered as the same value to every operator in the topology.

That would be the whole post, except that StoatFlow keeps two of those clocks, not one. It accepts a late record the way Kafka Streams does — against the operator's own stream time — and closes the window the way Flink does, on the watermark. Getting that right matters for a ported application, and it matters for us: our compatibility matrix said something different, and the first version of the correction we published was wrong too. This post sets the three clocks side by side, walks one arrival sequence through all three, and states exactly what a Kafka Streams application sees when it starts reading StoatFlow's.

TL;DR

• What: Three definitions of event-time "now". Kafka Streams — per task, the maximum record timestamp seen, optimistic. Flink — per operator, the minimum watermark over its inputs, pessimistic and distributed. StoatFlow — one global watermark, the minimum across non-idle partitions minus the out-of-orderness bound, computed at the edge.
• Why it matters: the clock decides when a window's final result is emitted, when suppress releases, when an event-time timer fires, and what currentStreamTimeMs() returns.
• How StoatFlow does it: Flink's WatermarkStrategy API per source, one generator per partition, one minimum on the dispatch path — and, inside every windowed operator, a Kafka Streams-style stream time that decides whether a late record is still accepted.
• The guarantee: every operator sees the same watermark in the same epoch, so there is no per-operator disagreement about what is late. A Kafka Streams application ported to StoatFlow counts the same records into each window; only the instant the final result appears changes.
• The catch: that instant is later — by the out-of-orderness bound at least, and by up to two commit intervals on top of it (part 3). One lagging, non-idle partition delays every window in the application. currentStreamTimeMs() returns the watermark, which is lower than Kafka Streams stream time and global rather than per-task.
• The correction: Kafka Streams grace has bounded acceptance, not retention, since KIP-328 — and StoatFlow's acceptance rule is the same one. The difference between the engines is the clock that closes the window, not the rule that admits a record.
• Docs: Event time and watermarks · Kafka Streams compatibility matrix.

In this post

• Three answers to "is this window done?"
• Kafka Streams: one clock, optimistic and per-task
• Flink: one clock, pessimistic and per-operator
• StoatFlow: both clocks
• Grace: the same rule — the close is what differs
• What a ported application sees
• Tradeoffs and limits
• What's next

Three answers to "is this window done?"

A windowed aggregation makes two decisions about time, and it is worth keeping them apart from the start. The first is acceptance: a record arrives for a window whose end has passed — is it folded in, or dropped? The second is the close: at what moment is the window's result final, so that suppress can release it, an onWindowClose emission can fire, and the state can eventually be expired? Each decision needs a clock, and the three engines pair the two decisions with clocks differently.

Kafka Streams uses one clock for both decisions; so does Flink. StoatFlow uses Kafka Streams' clock for the first and Flink's for the second. The clocks themselves differ along five axes:

Kafka Streams: one clock, optimistic and per-task

Kafka Streams' notion of event-time "now" is stream time: the largest record timestamp a task has processed so far. It lives as a plain field on the windowed processor — observedStreamTime, advanced with Math.max on every record — and since a processor instance is created per task, it is a per-task quantity. A task that owns three partitions has one stream time across them; a different task has another.

Both decisions hang off it. When a record arrives, KStreamWindowAggregate first advances stream time with the record's own timestamp, computes windowCloseTime = observedStreamTime − grace, and then folds the record into every matching window whose end is still greater than that close time; the rest are logged as expired and counted in the dropped-records sensor. Emission on window close — EmitStrategy.ON_WINDOW_CLOSE, or suppress(untilWindowCloses) — fires when the same clock passes the same threshold.

The clock is optimistic: the moment any record with a timestamp from the future arrives, stream time jumps to it, and every window whose close it has now passed stops accepting and emits its final result — even if a sibling partition still holds older records that belong in those windows. Kafka Streams knows this, and the mitigation is timestamp synchronisation: within a task, the next record to process is always the one with the lowest timestamp across the task's buffered partitions, and max.task.idle.ms (default 0) says how long to wait for a partition whose buffer is empty. At the default the task waits only for data already on the brokers, never for a producer that has yet to send. It is cheap — no coordination, no extra state — and the result is available the instant the last record is processed. The cost is that a single fast partition, or a single record stamped ahead of its neighbours, closes windows early. That is a reasonable trade for a fleet of independent tasks with no channel between them, and it is the model StoatFlow keeps for one of its two decisions.

Flink: one clock, pessimistic and per-operator

Flink's answer is the watermark: a monotonic claim, inserted at the source, that no record with a timestamp below T should be expected any more. Each source split runs its own generator — forBoundedOutOfOrderness(d) emits maxSeen − d − 1, forMonotonousTimestamps() emits maxSeen − 1, withIdleness lets a quiet split drop out of the calculation — and the watermark travels in-band as a special event through the operator graph. Every downstream operator takes the minimum over its input channels and forwards that as its own watermark.

Both of Flink's decisions run on it. A window fires when the operator's watermark passes the window end, keeps its state for allowedLateness, re-fires an updated result for each late record that lands inside that span, and drops a record only once the watermark has passed end + allowedLateness. The clock is pessimistic by construction — it lags the newest data by the out-of-orderness bound, and by the slowest input — which is exactly what makes a final result final.

The price is that "the watermark" is not one number. Two operators in the same job hold different watermarks at the same instant, because they sit at different depths and behind different inputs, so they can disagree about whether the same record is late. That per-operator divergence is the fourth of the failure modes the internal-consistency post discusses, and it is the reason Flink needs machinery around the clock: FLIP-217 alignment pauses a split that has run too far ahead of its group. That is a consequence of distribution, not of watermarks. One thing Flink does not do is checkpoint the watermark — timers survive a restore, the watermark does not (FLINK-5601, open since 2017) — so after a restore every operator's watermark restarts at Long.MIN_VALUE and is re-derived from what the sources emit next. StoatFlow re-derives its watermark the same way (part 3).

StoatFlow: both clocks

StoatFlow's public API is Flink's, deliberately. A source declares its time semantics with a WatermarkStrategy — forBoundedOutOfOrderness, forMonotonousTimestamps or noWatermarks, plus withTimestampAssigner and withIdleness — attached per source through Consumed.withWatermarkStrategy. A source that declares nothing gets forBoundedOutOfOrderness with the application-wide stoatflow.watermark.max-out-of-orderness-ms, 10 seconds by default. The withWatermarkAlignment decorator from FLIP-217 exists too, signature-identical, for code written against Flink.

val orders = builder.stream(
    "orders",
    Consumed.with(Serdes.String(), orderSerde)
        .withWatermarkStrategy(
            WatermarkStrategy.forBoundedOutOfOrderness<String, Order>(Duration.ofSeconds(30))
                .withTimestampAssigner { _, order, _ -> order.placedAt }
                .withIdleness(Duration.ofMinutes(2)),
        ),
)

What happens behind that API is where StoatFlow departs from both engines. One generator runs per source partition, and because the whole topology is one process, there is no operator graph to propagate through and no shuffle to cross. Each partition's generator advances as the dispatcher — the single thread that hands records to the key-affinity lanes — dispatches its records, and the tracker takes the minimum across all partitions that are not idle and stores one value: on the consumer thread every 200 ms, and once more just before every commit barrier. That value is delivered to every sub-topology as the same event, in the same epoch. Two operators cannot disagree about the watermark, because there is only one; there is no coordination protocol to configure, because there is nothing to coordinate. This is the property the internal-consistency result rests on, and it comes free with the single-instance architecture rather than being engineered on top of it.

That watermark is the close clock. onWindowClose emissions, suppress(untilWindowCloses), the null side of a left or outer stream-stream join, event-time timers and STREAM_TIME punctuators all fire on it.

The acceptance clock is a different value. Every windowed aggregation in StoatFlow also keeps a stream time of its own — the highest event time that operator has processed, an AtomicLong because records reach it from many lanes at once — and the late-drop rule reads that, not the watermark: a record is dropped when its window's end + grace is already behind the operator's stream time, and folded in otherwise. Session windows use Kafka Streams' post-merge predicate, so a late record that bridges two still-open sessions is rescued exactly as it would be there. Drops are counted in stoatflow.dropped.records.total with reason="late_window" or "late_session", the counterpart of Kafka Streams' dropped-records, and never sent to a dead-letter topic. Why not judge acceptance on the watermark too? Because it is pessimistic by the bound and by the slowest partition: every window would keep admitting records for that long after Kafka Streams had sealed them. The watermark's job is to make the final result safe, not to widen the window.

Grace: the same rule — the close is what differs

Our Kafka Streams compatibility matrix carried a row for grace that read: Kafka Streams — retention after close; StoatFlow — extends acceptance window. ADR-016, the decision record behind StoatFlow's window lifecycle, said the same thing in more words: Kafka Streams' grace "extends state retention, not event acceptance", and late records "are silently dropped unless using suppress". It needs correcting, twice.

First correction: Kafka Streams grace bounds acceptance. The retention description was true before KIP-328 landed in 2.1. Since then TimeWindows.ofSizeAndGrace(size, grace) is an acceptance bound in exactly the sense the ADR claimed for StoatFlow: a record is folded in while windowEnd > observedStreamTime − grace and dropped — logged, and metered — once it is not. Retention is a separate, storage-level knob (Materialized.withRetention, which must be at least size plus grace) on both engines.

Second correction: StoatFlow's acceptance rule is the same one. When we fixed the row after finding the first problem, we wrote that StoatFlow measures acceptance against the global watermark. It does not — the section above is the code. Both engines admit a record while windowEnd + grace is ahead of the operator's max-observed stream time; StoatFlow's comparison is strict where Kafka Streams' is inclusive, so a record whose window closes exactly at stream time is kept for one more tick, and that is the entire divergence on the way in. What the watermark governs is the way out: the moment the result is final.

The diagram makes the split concrete. One window from 10:00 to 10:05 with a minute of grace, so it closes at 10:06; two source partitions; a 30-second out-of-orderness bound; nine records in the order they reach the operator.

Read the two clock rows first; each shows the clock's value after that arrival has been folded in. Stream time — the highest event time seen — passes the 10:06 close at arrival ⑦, a record from the next window stamped 10:06:40. The watermark passes it only at arrival ⑨, because P1 is behind and the minimum follows the partition that is behind. Record ⑧, stamped 10:04:45 and squarely inside the window, arrives between those two moments.

Now the three verdict rows. Kafka Streams drops ⑧ — stream time is 10:06:40, the window's close was 10:06:00 — and has already emitted its final count of 5 at ⑦. StoatFlow drops ⑧ for the same reason, on the same clock, and emits the same final count of 5 — at ⑨, when the watermark passes the close. Flink's acceptance clock is the watermark, which reads 10:04:00 when ⑧ arrives (10:04:15 once it is counted), so Flink keeps it and emits 6 at ⑨. None of the three is wrong; StoatFlow keeps the Kafka Streams admission rule so a port behaves identically on the way in, and takes the later close so that "final" means every contributing record is behind one application-wide claim. The concept page's timeline walks the single-partition lifecycle of the same window.

What a ported application sees

A Kafka Streams application ported to StoatFlow keeps its window definitions, its grace periods and its suppress calls, and its windows admit the same records. Four things change.

currentStreamTimeMs() returns the watermark. In Kafka Streams, ProcessorContext.currentStreamTimeMs() is the task's max-observed timestamp. In StoatFlow it delegates to currentWatermarkMs(), the native name for the global watermark — the minimum across non-idle partitions, minus the out-of-orderness bound. Two consequences follow: the value is generally lower than Kafka Streams stream time, and it is global rather than per-task. Logic that compares a record's timestamp to "now" sees a larger gap; a deadline keyed on stream time fires later. Before the first record has been processed the watermark is Long.MIN_VALUE, and isLate() is defined against it. The porting guide flags this divergence first.

Final results arrive later. An onWindowClose emission, a suppress(untilWindowCloses) release, or the null side of a left join appears when the watermark passes the close — later by at least the out-of-orderness bound, plus up to two commit intervals under exactly-once (part 3). Results emitted on every update, the default, are unaffected.

STREAM_TIME punctuators fire on the watermark. Kafka Streams fires them on stream-time advance; StoatFlow on watermark advance. A punctuator that expected to run the instant a future-stamped record arrived now runs when the whole application has caught up to it.

One source can opt back into stream time. Consumed.withTimestampExtractor(extractor) adapts a Kafka Streams TimestampExtractor onto a per-source strategy with a monotonous generator, so that source contributes Kafka Streams stream time to the watermark and its windows close as they did before the port. It shares the single per-source strategy slot with withWatermarkStrategy; part 2 has the details, and the bug it shipped with.

Tradeoffs and limits

The close is conservative, always. Every window's final result is later than under Kafka Streams by at least the out-of-orderness bound, because the watermark trails the newest data by that much before the slowest partition is even considered. The 10-second default is a default, not a recommendation: size it to the disorder the data actually has, or the latency is paid for nothing.

One lagging, non-idle partition delays every window. The minimum has no locality. A partition that keeps delivering but delivers old timestamps holds the watermark for the whole application — every window, every timer, every suppress buffer — until it catches up or goes quiet long enough to be excluded. Part 2 is the configuration discipline that follows from that.

Two one-tick divergences at exact boundaries. The late-drop comparison is strict where Kafka Streams' is inclusive, so a record whose window closes exactly at stream time is kept one tick longer; and a left or outer stream-stream join opens its null side when the watermark reaches the close, where Kafka Streams waits for it to pass. Both are recorded in ADR-118 and both are deliberate — the first survived a review that proposed the inclusive form, because the inclusive form drops a boundary record that the exactly-once close (part 3) is designed to include. Output sets are identical; the boundary tick is not.

currentStreamTimeMs() is an alias. It is kept so ported code compiles, not because it returns the Kafka Streams quantity; there is no per-task stream time to ask for.

Ingestion time is not supported. Event time from the record or an assigner, or processing time; not the broker append time as a third notion.

DIRECT dispatch has no watermarks. The processing-time fast path skips watermark tracking, and the engine refuses to combine it with an event-time topology; nothing in this post applies to it.

What's next

The minimum is the sharp edge of this design. Everything above assumes the watermark moves — and a single partition that stops advancing, or a single source configured with noWatermarks(), holds every window in the application still. Part 2, Making the watermark move, is the discipline that follows: what each source has to say about its own time, what idleness actually does and the bug that made it necessary, why one strategy stalls a whole topology, and why StoatFlow's ordered dispatch already does most of what Flink needs watermark alignment for.

Read on:

• Event time and watermarks — the concept page this post extends.
• Kafka Streams compatibility matrix — the corrected time-semantics rows.
• Internal consistency — what one application-wide watermark makes possible.
• KIP-328: Ability to suppress updates for KTables — where Kafka Streams grace became an acceptance bound.
• KIP-695: Further improve Kafka Streams timestamp synchronization — max.task.idle.ms, and why an optimistic clock needs it.
• FLIP-217: Support watermark alignment of source splits — Flink's answer to per-operator drift.
• FLINK-5601: Window operator does not checkpoint watermarks — why a Flink watermark, like StoatFlow's, is re-derived after a restore.