Closing a window under exactly-once

The naive model of a windowed aggregation says the final result is emitted the moment the watermark passes windowEnd + grace. In StoatFlow it is not: the result is emitted at a commit barrier, and the watermark it is judged against is the one stamped on the previous barrier — so a final emission can trail the watermark by up to two commit intervals. That lag is deliberate. It is the price of a final result whose every contributing record is already committed on every lane, which is what "exactly-once" has to mean for a window rather than for a record. Part 2 got the watermark moving; this post follows a single advance from the consumer thread to the committed result.

TL;DR

• What: the path from a watermark advance to a committed window result — the punctuation lane, the pre-barrier force-emit and the barrier's stamp, emit strategies and suppress, the one-clock late rule, the committed-watermark frontier, the durable floor and emit frontier, and event-time store expiry.
• Why: under exactly-once a final result must not include a record the transaction can still roll back — and it must not be emitted while a record that will still be admitted is queued on another lane. The watermark says "no earlier records are coming"; it does not say "every earlier record is processed and committed". The frontier closes that gap.
• The guarantee: a window's final result is emitted exactly once, reflects every record the operator admitted, and survives a crash, an in-place restart and an HA promotion — the emit frontier and the late-drop floor are restored with the state, and the watermark resumes from the partition times committed with the offsets.
• The catch: final emissions trail the watermark by up to ~2 commit intervals — two more per sub-topology hop they are fed through — and so does the null side of a left/outer stream-stream join; a record is late once the watermark stamped on the last barrier its lane crossed has passed its window's close, which can count different records into a window than Kafka Streams for out-of-order or filtered input; a session closes only after its last event plus the gap plus grace.
• Docs: Windowing · Exactly-once · Event time and watermarks.

In this post

• What a watermark advance triggers
• Emit strategies and the three suppress triggers
• One clock decides late
• The committed-watermark frontier
• What survives a restart
• When state expires
• Tradeoffs and limits
• What's next

What a watermark advance triggers

The global watermark is recomputed in three places. The consumer thread asks for a new minimum after every poll, and the tracker emits one at most every 200 ms — stoatflow.watermark.auto-interval-ms, Flink's pipeline.auto-watermark-interval under another name. The dispatcher recomputes it once more immediately before it fixes a commit barrier — a force-emit — and stamps that value on the barrier, so the stamp covers every record the closing epoch dispatched. And shutdown recomputes it a final time, for a final barrier that carries a flag this post returns to.

An advance does not reach the operators as a number they read. It is delivered as an event — one per sub-topology — on that sub-topology's punctuation lane: the dedicated virtual thread that runs punctuators and, since ADR-084, watermark processing, with the same epoch isolation and state access as record processing. The commit barrier travels down the same lane, and the dispatcher enqueues the watermark event ahead of it. The lane is FIFO, so the advance is processed inside the epoch the barrier closes — but what closes windows is the stamp on the barrier, not the event.

What acts on the event itself is the part of the engine that fires on event time: event-time timers, STREAM_TIME punctuators and STREAM_TIME scheduled sources. Everything that closes something — every windowed aggregation, tumbling, hopping, sliding, session and their cogroup forms, both watermark-driven suppress processors, and the stream-stream join — acts at the next barrier instead, on the stamp. The event also marks the punctuation lane as having done work: the idle-throttled barrier cadence (ADR-134) treats an epoch as empty only when no record was dispatched, that lane did nothing, and no sub-topology still has final results to emit or commit. That last term is the emission work signal (ADR-149): under at-least-once a downstream sub-topology's input is its upstream's final results, forwarded in memory and never sent, so without it the last windows would wait for new input once input stops. With it, barriers keep coming until those windows are emitted and committed.

Then the twist this post is about. For anything that closes — onWindowClose, suppress(untilWindowCloses) and untilTimeLimit, the null side of a join — the watermark event emits nothing. That waits for the barrier.

Emit strategies and the three suppress triggers

A windowed aggregation decides when to forward with an EmitStrategy. onWindowUpdate(), the default, forwards on every record, and none of what follows applies to it. onWindowClose() forwards nothing per record: when the frontier passes a window's close, it range-fetches the store for every window whose close lies between the last close it already emitted and the frontier, forwards each exactly once, and advances that emit frontier. The cost is latency — at least the grace period, plus the frontier lag below. Kafka Streams' EmitStrategy.onWindowClose() does the same on each aggregation's own stream time.

suppress is the other way to hold a result, and it is worth one table, because the three triggers run on two different clocks:

The two watermark-driven buffers are the Kafka Streams ones on a different clock; both span epochs, so both are backed by a changelogged store. untilBarrier is StoatFlow's own, and it is not a time trigger at all. It buffers the records that flow through it during an epoch and releases at most one per key at the commit barrier, inside that epoch's transaction, carrying the operator's state as of the epoch's consistent cut. Nothing is emitted between barriers. It needs no store and no changelog: the buffer lives inside one epoch, and on a crash the transaction aborts and the epoch is reprocessed, rebuilding it. That is the valve the internal-consistency post is built on — a coherent answer at every commit rather than a stream of partial ones — and, unlike an eager untilTimeLimit buffer, it never evicts an intermediate under memory pressure, because an evicted intermediate is precisely the inconsistent answer it exists to prevent.

Both carry the buffered record out: a release keeps the latest update's headers and timestamp, and records released together leave in buffer-time, then key, order — as in Kafka Streams. BufferConfig.maxBytes(...) bounds the serialised buffer.

One clock decides late

"Late" happens twice on a record's way into a window, and only the second one drops anything.

The first is at the dispatcher. After a batch has advanced the per-partition watermarks, every record whose event time is below the global watermark is counted as a late arrival — stoatflow.watermark.late.records.total, a debug-level meter — and then dispatched like any other. Nothing at this layer drops a record or sends it to a dead-letter topic; the counter measures how out-of-order the input is, not how much of it is lost.

The second is inside the windowed processor, and it judges the record's window, not the record. Every lane carries a floor: the watermark stamped on the last commit barrier it crossed. A record is dropped when its window closes at or before that floor — end + grace for a time window, start + D + 1 + grace for a sliding window, mergedEnd + gap + grace for a session. Every window kind and emit strategy uses this one test, and so does suppress(untilWindowCloses); it is the clock that closes the window too, so a window that has closed never changes again. Those drops are the real signal: stoatflow.dropped.records.total{reason="late_window"} (late_session for sessions), Kafka Streams' dropped-records parity, at info level. We used to admit on a per-operator stream time and close on the watermark, as part 1 described; behind a filter the two disagreed, and a window already emitted could still take a write. Since rc.4 the watermark is the only clock (ADR-149). Kafka Streams decides on each aggregation's own clock, the largest event time that reached it, so StoatFlow admits a record up to the out-of-orderness bound more out of order and, behind a filter, drops one Kafka Streams would still count.

The boundary is inclusive: a window whose close time equals the floor is closed — as the emission is, and as Kafka Streams' drop is. It used to be strict, one tick longer than Kafka Streams, to protect the race the next section is about: the watermark advanced, on other keys, to exactly a window's close while a record for that window is still queued on its lane. With the floor taken from barrier stamps that record cannot lose: it was dispatched before the barrier whose stamp reaches the close, so its lane judges it against the stamp before, and admits it. The one-tick entry in the ADR-118 register is retired.

Sessions are judged after the merge scan, on the merged session: a record is late when mergedEnd + gap + grace is at or before the floor, and is counted as late_session. A late record that bridges two still-open sessions is folded in and rescued, as in Kafka Streams. A closed session, though, is final: it is left out of the merge scan, so an on-time record within its gap starts a new session instead of re-opening one already emitted — Flink's behaviour; Kafka Streams merges and retracts.

The committed-watermark frontier

The watermark makes one claim: no record with a lower timestamp is still to come from the sources. It does not claim that every record with a lower timestamp has been processed, and inside a single instance that gap is concrete. Records reach a windowed operator from many key-affinity lanes at once, each with its own queue.

Take a window [10:00, 10:05) with a minute of grace, closing at 10:06:00, on a source with a 30-second out-of-orderness bound. A record R stamped 10:04:50 — inside the window, inside grace — is dispatched to lane 7 and queues behind other keys. A moment later a record F stamped 10:06:30 is dispatched to lane 3, and because the watermark is the maximum seen less 30 seconds, F moves the raw watermark to 10:06:00: the window's close. Under the naive model the punctuation lane now processes the watermark event, range-fetches the window and emits its final count. R is not in it.

Then lane 7 processes R. Is R admitted? Judge it against the live watermark and the answer depends on timing: R was on time when it was dispatched, but a lane that reaches it after F has moved the watermark sees a window that is already closing. Admit it, and the final count already emitted is wrong, permanently, because onWindowClose emits each window once; drop it, and a record that arrived in time is lost. Same input, two outcomes, decided by which virtual thread the scheduler ran first — and the emission happened before either had run, so it could not know. Under exactly-once there is a second problem on top: the records that did contribute to that count are in a transaction that has not committed and can still abort.

StoatFlow's answer is to judge both the drop and the emission against watermarks the barriers carry, never the live one. At offset fixation the dispatcher stamps the current watermark on the barrier — call barrier N's stamp W(N) — and writes it to the state, where it commits with transaction N. Every lane keeps a short ring of the stamps it has crossed:

floor of a record lane that crossed barrier N   = W(N)
frontier of the punctuation lane at barrier N   = W(N−1)

A record lane judges every record it processes next against its floor. The punctuation lane, once every record lane of its sub-topology has crossed barrier N, hands W(N−1) — the stamp of the barrier before — to every watermark-aware operator of the sub-topology as the gate for its final emissions, if it moved: the range fetch above runs up to it, not up to the raw watermark. The emitted records carry the event time of what they emit — a time or sliding window its latest record's time, a session its end, a suppress release its buffered record's time, the null side of a join its unmatched record's time — as in Kafka Streams, so a final result windowed again downstream lands in the right window. Only the moment they appear depends on the barrier.

Why the frontier is safe to emit against comes down to two facts about barriers. First, barrier N is only ever created after epoch N−1's transaction has committed — the "one committing, one processing" rule of ADR-077 — and every record dispatched before barrier N−1 was processed before its lane reported that barrier, because lanes are FIFO. So at barrier N, everything dispatched before N−1 is not just processed but durable. Second, every record lane that has crossed barrier N−1 judges what it processes next against W(N−1), so a record for a window closing at or below W(N−1) that comes later is dropped as late. Put together: at barrier N, the membership of every window closing at or below W(N−1) is settled — no record can still join it — and committed — every record that did join it is in a transaction that cannot abort. In the scenario, R and F were both dispatched in epoch 3, and barrier 3 was stamped 10:06:00. Lane 7 processes R before it crosses barrier 3, against its floor W(2) = 10:05:20, so R is admitted whichever lane runs first; both are committed in transaction 3, the frontier reaches 10:06:00 at barrier 4, and the count emitted there is the one that actually happened.

The emission rides the barrier, not the watermark event, for a reason that is easy to miss: the frontier only changes at barriers, and an idle stream fires no further watermark events, so the barrier is the only tick guaranteed to keep arriving — and the emission work signal keeps barriers coming until the last final results are emitted and committed, under at-least-once too, where nothing else would. It is also what makes the flush exactly-once. The lane has already reported barrier N when it flushes, so the emitted records are epoch-gated into transaction N+1, together with the update to the durable frontier in the next section — the result and the record that it has been emitted commit atomically.

That is where the lag comes from. A watermark advance that lands just after barrier N−1 is fixed is stamped on barrier N and becomes the frontier at barrier N+1 — emitted at N+1, up to two commit intervals after the advance. A sub-topology fed by another's final results — a window over a window across a re-key, say — runs two barriers further back for each such hop, because those finals reach it one epoch after they were emitted. The interval self-tunes: stoatflow.commit-barrier.interval-ms seeds it at 500 ms, interval-factor (1.5) scales it with the measured commit duration, and min-interval-ms / max-interval-ms clamp it to [150 ms, 5 s] by default. So the frontier adds between 0.3 and 10 seconds to a final result, typically around a second, and as much again per such hop, on top of the out-of-orderness bound Part 1 already charged.

Read the frontier rows against the barriers: barrier 3 is stamped 10:06:00 while the frontier is still barrier 2's stamp, 10:05:20, so nothing is emitted; at barrier 4 the frontier becomes 10:06:00 and StoatFlow emits, into transaction 5 — after R and F, on the lane rows, have both been processed and made durable by transaction 3. Lane 7 judged R against barrier 2's stamp, 10:05:20, so it was admitted whichever lane ran first.

Two smaller mechanics complete the picture. Shutdown does not wait for two more barriers: the final barrier carries a final flush. Each punctuation lane waits until every record lane of its sub-topology has crossed it — every record of the last epoch processed — then emits every window closing at or before the barrier's stamp before it reports the barrier, so those results commit in the final transaction, in every sub-topology, and the barrier records that it flushed, so the next start resumes at that watermark. A record lane that does not cross in time aborts the barrier instead, and the next start resumes from the cut before. And an unset frontier — Long.MIN_VALUE, before the first stamps have rolled through — runs no emission pass at all.

What survives a restart

The floor is persisted, with the state. The barrier stamps do not live only in the engine's memory: the dispatcher writes the last few into an engine store, stoatflow-watermark — in memory, changelogged to <application.id>-stoatflow-watermark-changelog, one compacted partition — before the epoch rotates, so they commit inside the barrier's own transaction, exactly with the cut. A crash restart, an in-place engine restart (ADR-127 builds a fresh engine) and an HA promotion all seed every lane's floor and every punctuation lane's frontier from it. After a crash the newest stamp is left out: the emission pass barrier c triggered commits only with transaction c + 1, so the first barrier after the restart plays barrier c's role and runs that pass again, judged against the durable emit frontier below. After a graceful shutdown it is kept: the final barrier flushed everything up to its stamp and recorded that it did. The watermark itself resumes from each source partition's largest event time, committed with the offsets in Kafka Streams' TopicPartitionMetadata format, not from re-read records.

Why the state and not the offsets: offset tools — kafka-consumer-groups --reset-offsets, MirrorMaker 2's offset sync — rewrite offsets with empty metadata, and offsets expire after a long downtime; none of that touches a changelogged store. Flink does not checkpoint watermarks (FLINK-5601, part 1), and a Kafka Streams windowed aggregation rebuilds its stream time from the records it processes after a restart. What this buys is that no record late before a restart is admitted after it, and the first window close after a restart does not wait for event time to be rebuilt.

The emit frontier is persisted. Each onWindowClose operator keeps the highest close it has already emitted in a companion store — <store>-emitfrontier, a compacted key-value store registered like any other, so it rides the ordinary changelog write, the epoch commit and the restoration path with no bespoke code of its own (ADR-130). Because it is an ordinary store, the frontier update and the emitted results commit in the same transaction, and a window emitted in an epoch that later aborts is emitted again, once, when the epoch is reprocessed — never twice, and never zero times. Four integration tests pin it: a pending window or session is re-emitted exactly once after a crash, an already-emitted one is not re-emitted, and a session that merges across a restart emits its merged result once.

The same idea appears at a Kafka Streams cutover. With the frontier companion left unseeded, the first emission pass after the migration tool has translated a windowed changelog would re-emit a final result for every restored closed window still inside retention, in one burst; emitFrontier: true seeds the companion from the store's own maximum changelog timestamp, so nothing already emitted is emitted again. The late-drop floor, which a migrated application does not have yet, is taken from Kafka Streams' own partition times in the committed offsets.

When state expires

Windowed, sliding, session and cogroup stores expire their state at their retention (Materialized.withRetention, else size + grace, 2·D + grace or gap + grace), and the clock that drives it is never the raw watermark:

Sessions used to be the odd row, expiring on the operator's own stream time; with that clock gone they expire on the committed watermark like everything else. The drop, the close and the expiry read one clock, and no lane's floor is behind the watermark expiry uses — the punctuation lane runs its pass only once every record lane has crossed the barrier — so nothing admitted lands in a window or session already removed.

There is one more schedule, and it is not event time at all. The store-level retention tick that reclaims versioned stores' history (ADR-142) fires from the punctuation lane's barrier handler, after the lane has reported the barrier and after the frontier flush — not from a watermark event. Three reasons, in the order the code gives them: the deletes become ordinary epoch mutations, so a local delete and the source offsets commit in one transaction rather than best-effort; the epoch attribution is right, because the handler has already cut the epoch; and it fires for every topology, where watermark events only exist in event-time ones.

What all of this means for state size under onWindowClose(): effective retention is bounded by retention + segmentInterval + (watermark − emit frontier), and that last term is the frontier lag from the previous section. The windowing page's retention contract has the regimes and worked examples.

Tradeoffs and limits

Final results trail the watermark by up to two commit intervals — two more for each sub-topology hop they are fed through. With the default cadence bounds that is between 0.3 and 10 seconds, typically about a second, on top of the out-of-orderness bound. Compute yours from stoatflow.commit-barrier.min-interval-ms and max-interval-ms; a lower max-interval-ms tightens the worst case at the cost of more commits. onWindowUpdate() results are unaffected — they forward per record and never wait for a barrier.

The null side of a left or outer stream-stream join trails the same way, because it is a final emission gated on the same committed watermark. One further divergence rides with it: N identical unmatched arrivals at the same key and timestamp emit N null-joins, where Kafka Streams' plain put overwrites and emits one. Output sets are identical; timing and multiplicity are not.

Late records are dropped silently at the processor. A debug log line and the dropped.records.total counter — no dead-letter topic, no callback. If the record matters, widen grace, or intercept it upstream in a Processor with isLate(), which is true when the record's timestamp is below the watermark during process(...).

Lateness can differ from Kafka Streams. Kafka Streams decides per aggregation, on the largest event time that reached it; StoatFlow decides on the watermark, which trails the newest event by the out-of-orderness bound and also moves on records a filter removed. A record up to that bound out of order is admitted where Kafka Streams drops it, and behind a filter one Kafka Streams would still count is dropped (KSC-132 in the ADR-118 register).

A crash restart re-runs one emission pass. The floor and the emit frontier come back with the state, so nothing is emitted twice and nothing late before the crash is admitted after it; the pass the last committed barrier triggered had not committed, so the first barrier after the restart runs it again. With stoatflow.changelog.enabled: false the floor has no changelog to come back from, and the engine resumes it from the partition times committed with the offsets.

untilBarrier is a valve, not a trigger. It consults no clock and can be combined with a time-based suppress upstream, but it is not a substitute for one.

One extra changelog topic per onWindowClose aggregation, and one per application. The emit frontier's companion store is compacted and takes its aggregation store's changelog partition count; it appears in the topic list and the restore plan. Every topology with a windowed aggregation, a time-based suppress or a stream-stream join also gets <application.id>-stoatflow-watermark-changelog, one compacted partition, for the floor; it is registered at engine start, so describe() does not list it, and a deployment that pre-creates its topics must create it too.

What's next

Everything above is observable — but only if you know which counter is off by default, why a window-close test that passes in the TopologyTestDriver emits later on a real broker, and what /watermarks is telling you when the global value stops moving. Part 4, Testing and operating event time, is the operator's half of this series: which tests transfer, how to read the endpoint, the meters that matter, the one alert to set on day one, and what a restart does to the watermark.

Read on:

• Windowing — emit strategies, suppression, buffer configuration, the retention contract.
• Exactly-once — the commit barrier the frontier hangs off.
• Event time and watermarks — the concept page.
• Internal consistency — untilBarrier in action.
• Making the watermark move — part 2 of this series.
• Testing and operating event time — part 4 of this series.
• Migrating with data — the emitFrontier cutover flag.