Making the watermark move: sources, idleness, and the partition that stopped the world

In July, a partition that had never received a single record pinned StoatFlow's global watermark at Long.MIN_VALUE for the whole instance: every window in the topology stayed open, every event-time timer stayed armed, and idleness — the mechanism that exists for exactly this case — could not release it, because the idle marker required at least one event before it would consider a partition quiet. The fix was one line and a deletion. It is the right place to start, because it is the shape of every event-time problem in a single-instance engine. Part 1 established that StoatFlow's watermark is the minimum across non-idle partitions. A minimum has no locality: one source decides for all of them. This post is the configuration discipline that follows.

TL;DR

• What: per-source watermark configuration — forBoundedOutOfOrderness, forMonotonousTimestamps, noWatermarks, withIdleness, withTimestampExtractor — and what each does to the one number every window waits on.
• Why: the global watermark is min across non-idle partitions. A quiet partition or a bulk reload stamped in the past holds every window and timer in the application.
• How: idleness excludes a partition that stops delivering (5 minutes by default, per source with withIdleness); noWatermarks() is refused in a topology with event-time operators; withTimestampExtractor gives one source a Kafka Streams-style watermark; ordered dispatch serves the oldest partition head first, each partition in offset order, so a partition that is ahead waits in its buffer rather than being processed ahead.
• The guarantee: the watermark is monotonic and computed only from partitions that are delivering — an idle partition never holds it back, and a live-but-lagging partition is never mistaken for an idle one.
• The catch: idleness is wall-clock, so a genuinely slow source looks idle; a partition that wakes from idle delivers records the watermark has already passed; a bulk reload stamped at load time still starves the live side today.
• The fix this post shipped: per-source withIdleness(...) was accepted, validated and never applied — only the global key took effect. It now does what it says.
• Docs: Event time and watermarks · Timestamp extraction and watermarks · Core configuration.

In this post

• One minimum, N sources
• Idleness: the escape hatch
• The noWatermarks() trap
• A Kafka Streams-style watermark for one source
• Ordered dispatch: alignment that is always on
• Alignment: the knob you mostly won't need
• Tradeoffs and limits
• What's next

One minimum, N sources

Every source partition has its own watermark generator. For the default strategy that generator emits the highest event time it has seen, less the out-of-orderness bound; a monotonous strategy emits the highest event time less one millisecond; noWatermarks() emits Long.MIN_VALUE. On each pass — every 200 ms on the consumer thread, and once more just before every commit barrier — the tracker takes the minimum over the partitions that are not idle and stores it as the watermark, the one value Part 1 showed being delivered identically to every operator.

The arithmetic is unforgiving. Three partitions whose watermarks read 10:06, 10:06 and 10:01 give an application at 10:01. A window ending at 10:05 with a minute of grace is done at 10:06 on two of them and open on the third, and the third wins. Nothing about the busy partitions' progress reaches the windows until the slow one catches up, or is excluded.

GET /watermarks is how you see which partition is the minimum. Trimmed to two partitions, with one obviously behind:

{
  "globalMs": 1790071260000,
  "globalIso": "2026-09-22T10:01:00Z",
  "alignmentGroupMins": {},
  "alignmentBinding": null,
  "partitions": {
    "orders:0": {
      "watermarkMs": 1790071560000, "watermarkIso": "2026-09-22T10:06:00Z",
      "idle": false, "lastEventTimeMs": 1790071592000, "lastEventTimeIso": "2026-09-22T10:06:32Z",
      "alignmentGroup": null, "alignmentHeld": false, "driftFromGroupMinMs": 0,
      "idlenessTimeoutMs": 300000
    },
    "orders:1": {
      "watermarkMs": 1790071260000, "watermarkIso": "2026-09-22T10:01:00Z",
      "idle": false, "lastEventTimeMs": 1790071590000, "lastEventTimeIso": "2026-09-22T10:06:30Z",
      "alignmentGroup": null, "alignmentHeld": false, "driftFromGroupMinMs": 0,
      "idlenessTimeoutMs": 300000
    }
  }
}

globalMs equals the lower watermarkMs; that partition is the minimum. Its lastEventTimeMs is fresh, so it is not quiet — it is delivering old timestamps, which is a different problem from the one idleness solves, and one this post returns to. The alignment* fields are present whether or not you use alignment; without it they read null, false and 0, and alignmentBinding is null. idlenessTimeoutMs is the idleness timeout actually in force for that partition — the source's withIdleness if it set one, otherwise the global key — which is what makes idle and lastEventTimeMs interpretable once timeouts differ per source. A partition whose next record waits behind older ones elsewhere also shows previewWatermarkMs, previewWatermarkIso and headHeldSinceMs — the case at the end of the next section.

Idleness: the escape hatch

A partition that stops delivering would hold that minimum forever. Idleness is the release: once a partition has had no event for longer than its timeout, it is marked idle and excluded from the minimum until its next record; the moment one arrives it is included again. The timeout is a wall-clock duration — the tracker stamps a liveness clock on every event and compares it to now — and the default is five minutes, from stoatflow.watermark.idleness-timeout-ms.

Read the global row. P2 is the slowest partition, so it is the minimum while it is live. At 10:07 on the wall clock it delivers its last record, and its watermark stops at 10:04. For the next five minutes P0 and P1 keep advancing and the watermark does not move: P2 is quiet but not yet idle, and it still counts. At 10:12 the timeout elapses, P2 is excluded, and the watermark jumps six minutes to the lower of P0 and P1 — every window that had been waiting closes at once. At 10:15 P2 delivers again, a record stamped 10:06; its own watermark moves to 10:05 and it is re-included immediately. It is the minimum again, and it is far below the global value, which does not move down — the watermark is monotonic — and does not move up either, because P2 is back in the minimum and behind. It stays held until P2 catches up or goes quiet again.

The dashed row is the same scenario with the timeout at zero. Idleness is off, P2 always counts, and the watermark never rises above P2's own value — 10:04 for as long as it is quiet. No window in the application closes; no store expires. The engine logs one warning at start-up when a source is left on a zero global key, because nothing at runtime can tell this apart from a healthy topology whose partitions are all live.

The bug that made the clock start at registration. Until July, the idle marker considered a partition only after it had received at least one event — a hasReceivedEvent gate. A partition that never received a record could therefore never become idle, yet its generator's initial Long.MIN_VALUE sat in the minimum. The trigger is ordinary: more partitions than active keys, a low-traffic topic, a partition assigned and simply empty. The fix (2026-07-04) seeds the liveness clock at registration, so a silent partition idles out after the timeout like any other, and deletes the gate from the idle-marking loop (the field itself survives — alignment's group minimum still reads it). One line and a deletion. Flink starts its idleness clock at registration for the same reason.

Why you want the built-in and not a filter of your own. The idle marker runs on a different thread from the one that records events, so a stale read could mark a live-but-lagging partition idle — and the watermark, being monotonic, would lock in the over-advance and drop the records in the gap as late. The exclusion therefore re-checks the liveness clock itself, reading it in the mirror of the writer's order, so a fresh watermark is never observed with a stale clock. A genuinely idle partition is still excluded; a racing one never is. That is the reasoning an ad-hoc "skip quiet partitions" filter does not get for free.

The second defect. WatermarkStrategy.withIdleness(Duration) is part of the per-source API, and both the concept page and the compatibility matrix described it as per-source idleness. Writing this section against the code showed that every strategy stored and validated the value and nothing ever read it: the engine registered each partition with the generator and the alignment config, and the tracker was built once with the global key. A source that asked for two minutes idled after the global five; with the global key at zero, it never idled at all. The fix landed with this post — each partition now carries its own effective timeout, the source's withIdleness if set, otherwise the global key. It shipped in 1.0.0-rc.3; on an earlier build only the global key takes effect. The first cut reached the DSL strategies but not the Processor API's extractor adapter; both are covered now, and a Topology.addSource(WatermarkStrategy, …) overload lets the Processor API set a strategy directly.

stoatflow:
  watermark:
    max-out-of-orderness-ms: 10000   # default strategy lag, 10 s — size it to the data
    idleness-timeout-ms: 300000      # 5 min; the default for sources without withIdleness; 0 disables it for them

// This source idles out after two minutes of silence, whatever the global key says.
Consumed.with(Serdes.String(), orderSerde)
    .withWatermarkStrategy(
        WatermarkStrategy.forBoundedOutOfOrderness<String, Order>(Duration.ofSeconds(30))
            .withIdleness(Duration.ofMinutes(2)),
    )

Not every stalled partition is a quiet one. A partition leaves its buffer in offset order, and the dispatcher serves the partition whose next record is oldest. So a record stamped far in the future — a producer clock that runs ahead, a timestamp in the wrong unit — waits at the head of its partition behind every older record elsewhere, and its partition dispatches nothing. That partition is not quiet, so idleness would release it only after the full timeout. It does not have to: with forBoundedOutOfOrderness and forMonotonousTimestamps, a partition whose next record has waited behind older ones counts towards the minimum with the watermark it will have once that record is processed, so the global watermark follows the other partitions within about a second, and the record keeps its place. stoatflow.watermark.preview.partitions counts such partitions. A custom WatermarkGenerator gets no preview: there the partition holds event time until the record goes or idleness excludes it. Either way, StoatFlow logs a warning naming the partition when its next record is stamped more than a minute ahead of the wall clock, and the warning says which case applies. Why a stream processor must never reorder a partition has the full story.

The noWatermarks() trap

noWatermarks() is documented as the strategy for processing-time-only topologies, and for those it is right: the watermark stays at Long.MIN_VALUE, and nothing in the topology waits on it. The trap is using it for one source among several, in a topology that has any event-time operator.

Two things would happen. First, setting any strategy on a source is an explicit event-time declaration, and the engine turns watermark tracking on for the whole topology when it sees one — noWatermarks() included. It does not opt the topology out of event time; it opts it in. Second, that source's partitions now sit in the minimum at Long.MIN_VALUE, and as long as they keep delivering records their liveness clock stays fresh, so idleness never excludes them. The global watermark is pinned for as long as the source is busy. Every window, every suppress, every event-time timer in the application waits on a source that has declared it will never say anything about time.

A noWatermarks() source that goes quiet for five minutes is idled out like any other, so the stall bites only while the source is healthy — which is why StoatFlow refuses the combination outright. A topology that reads a source with noWatermarks() and has a windowed aggregation, suppress, a stream–stream join, a processor that declares event-time timers or a STREAM_TIME scheduled source fails to build with a TopologyValidationException naming both, under StoatFlow.start() and the TopologyTestDriver alike. What the build cannot see is a STREAM_TIME punctuator, which is scheduled at run time: beside a busy noWatermarks() source it never fires.

The rule: a source with no meaningful event timestamps, inside an event-time topology, should use forMonotonousTimestamps() on the record timestamp. Its watermark then advances with its own records, and when it stops, idleness releases it. Reserve noWatermarks() for topologies with no event-time operators at all.

A Kafka Streams-style watermark for one source

Consumed.withTimestampExtractor(extractor) looks like a timestamp hook. It is more than that: StoatFlow adapts the Kafka Streams TimestampExtractor onto a per-source strategy whose generator is monotonous — the highest timestamp seen, less one millisecond — so that source's contribution to the watermark follows its newest record with no slack, as Kafka Streams' partition time does. The extractor receives the real consumer record, as it does in Kafka Streams, so an extractor that reads headers or partition metadata ports unchanged. Use it where a source's watermark should follow its newest record. Windows still close on the global watermark — the minimum across partitions — not on a per-aggregation stream time, so with several partitions they close once the slowest has passed them.

It shares the single per-source strategy slot with withWatermarkStrategy. Setting the extractor when a strategy is already present throws — "withTimestampExtractor cannot be combined with an already-set watermark strategy on Consumed (they share the single per-source strategy slot); use one or the other." — while the richer withWatermarkStrategy quietly wins if called second. Before this adapter emitted monotonous watermarks it emitted none, so a ported extractor silently never closed a window. That shipped, and was fixed.

Ordered dispatch: alignment that is always on

Flink users arrive with a specific fear. In Flink each source split is processed independently, so a split whose event time runs ahead feeds its records into window state while the operator's watermark — the minimum over inputs — waits on a slow split. The state of every window between "where the slow split is" and "where the fast split is" accumulates without bound. That is the failure mode watermark alignment (FLIP-217) was built to throttle.

StoatFlow does not have it, and the reason is structural rather than a feature. Polled records wait, still serialised, in a buffer per partition, and the consumer pauses a partition once 2000 are waiting — stoatflow.processing.buffered-records-per-partition, a pause threshold, so a partition peaks at that value less one plus max.poll.records. The dispatcher, the single thread that hands records to the key-affinity lanes, does not drain those buffers round-robin: it rebuilds a min-heap from the buffer heads and serves the partition whose next record is oldest, every time. Within a partition, records leave in offset order: a record older than the ones before it is not moved forward — it is out-of-order data, and the watermark strategy and the window's grace decide whether it still counts. A partition whose records run ahead in time therefore waits in its buffer; when that buffer fills, the consumer pauses the partition until the dispatcher has caught up on the others. The ahead partition is never processed ahead. Window state stays bounded, and the fast source is throttled, without any watermark-specific machinery. The ADR that introduced alignment puts it in one sentence: global-ordered dispatch is, in effect, alignment that is always on, bounded by buffer capacity.

The ordering has a precondition, and this is where Kafka Streams' max.task.idle.ms survives into StoatFlow as stoatflow.processing.max-task-idle-ms. Ordering across partitions is only as good as the buffers: if one buffer is empty because the poll has not fetched from that partition yet, the heap cannot see its next record, and "the oldest head" would be wrong. So before the dispatcher is signalled, the consumer thread asks, for each empty buffer, whether the broker still holds records for it. If it does, the whole dispatcher waits for the fetch. If the partition is caught up — lag zero — the default (0) proceeds; a positive value waits that much longer for a producer that might still send; -1 never waits. The check runs only when the topology tracks event time; a processing-time topology skips it. It gates all dispatch, not a single partition, and it is the one Kafka Streams timing knob that ports with its meaning intact.

Alignment: the knob you mostly won't need

WatermarkStrategy.withWatermarkAlignment(group, maxAllowedWatermarkDrift[, updateInterval]) is signature-identical to Flink's, down to the one-second default update interval. It works: partitions in the same named group are compared against each other, and a partition more than the drift bound ahead of a group member is held. A held partition keeps being fetched; the hold is in the dispatcher's ordering, not at the consumer. When the held partition's next record is the oldest one waiting anywhere and a lagging partition has nothing buffered, the whole application waits — nothing may overtake the oldest record — until that partition delivers or is declared idle; once every lagging partition has data, the record goes ahead in order. A partition with records waiting is never treated as gone from its group, while an empty one that stays silent idles out and releases the hold.

We built it for Flink-API parity, and our own design review then reframed the motivation. Because ordered dispatch already keeps the dispatch-time watermarks aligned, alignment is inert in the common fast-versus-slow-partition case: the fast partition's records buffer, the capacity pause handles it, and any hold yields, because the slow partition has data. It binds only for a sparse or slow-but-not-idle source — one whose buffer keeps running empty at zero broker lag while it remains non-idle and keeps counting toward the group minimum. Size the drift to cover that source's normal gap between records plus the difference in out-of-orderness bounds (drift ≥ G + b(sparse) − b(held); with equal bounds, the gap itself), or each gap stops the application. Every aligned source also needs an idleness timeout: without one, StoatFlow refuses to build the topology (alignment-idleness-off), because a silent laggard would hold for good. What it adds in that case is real: a deterministic event-time bound on the spread, where buffer capacity gives only a record-count bound that converts to a variable event-time spread depending on how dense the data is. Reach for it when a sparse ahead-source feeds a tight-grace stream-stream join and you need the skew capped in seconds rather than in records. Otherwise, the architecture has already done the work.

Alignment is code-only, per Consumed; there are no YAML keys for it.

Tradeoffs and limits

Idleness is wall-clock. A genuinely slow source — one record every six minutes — is idle under the five-minute default, and its records arrive behind a watermark that moved on without it. Set the timeout above the source's real inter-arrival time, or cover the gap with out-of-orderness. Nothing in the engine can tell a slow source from a dead one.

A woken partition is late by construction. Re-inclusion is immediate, but the watermark is monotonic: it does not come back down to meet the returning partition, and it does not move up again until that partition catches up. Records the partition delivers in the meantime are behind the watermark. Each is admitted while its window is still open — its close time (window end plus grace; for a session, its last event plus the gap plus grace) is still ahead of the watermark the processing lane holds — and dropped as late once it is not. A result already emitted as final is never revised.

A zero timeout means pinned, not fast. idleness-timeout-ms: 0 — now: for sources that set no withIdleness of their own — switches the escape hatch off. A partition that goes quiet holds the watermark until it speaks again; windows do not close; window and session stores never expire, and segmented stores accumulate RocksDB instances without bound. The start-up warning names this; it fires whenever any source is left on a zero global key — even when every partition is live, because there is no runtime signal for it — and stays silent when every source declares its own withIdleness.

noWatermarks() is refused beside anything the watermark drives — except a STREAM_TIME punctuator, which the build cannot see.

The bulk reload is not solved. Ordered dispatch is correct, and correctness has a cost: an old backlog's head wins every comparison. Picture three sources — live orders, live payments, and a customer master topic that is empty for 23 hours a day and then receives a hundred million records stamped with one load timestamp. That load timestamp is in the past, so it is the oldest head, so the dispatcher drains the entire reload before the next live record, and the watermark stands still for the duration — it cannot fall back to the load timestamp, and it cannot pass the reload partition until that partition has drained and idled out. Every live window waits. No watermark strategy fixes this: a bound or an idleness timeout does not change which partition goes first, only the timestamp does, and restamping the reload with live time would falsify its event time. The preview of a held head does not help either — it frees the watermark from a partition waiting behind older heads, and here those are the live ones. Alignment is the wrong tool too: it holds the source that is ahead; here the source that is behind is the problem. The inverse control — a dispatch tier that lets a background source defer to live traffic — is designed and on the roadmap, not shipped.

DIRECT dispatch has no watermarks, and the engine refuses the combination. The processing-time fast path skips watermark tracking; a topology that tracks event time cannot select it, so nothing in this post applies to it.

What's next

With every source configured so that the minimum moves, the question becomes what a watermark advance actually sets in motion — and under exactly-once the answer is not "the window emits when the watermark passes end plus grace". Part 3, Closing a window under exactly-once, follows one advance from the consumer thread to the committed result — and to the committed-watermark frontier that makes a final result trail the watermark by up to two commit intervals so that it can be provably final.

Read on:

• Event time and watermarks — the concept page, with the new noWatermarks() callout.
• Timestamp extraction and watermarks — the per-source API.
• Core configuration — the watermark keys.
• Architecture — data flow — ordered dispatch, in a sentence.
• REST API — state and progress — /watermarks.
• Roadmap — dispatch tiers.
• Three clocks — part 1 of this series.
• Closing a window under exactly-once and Testing and operating event time — parts 3 and 4.
• Why a stream processor must never reorder a partition — offset order, and the record stamped in the future.
• FLIP-217: Support watermark alignment of source splits
• KIP-695: Further improve Kafka Streams timestamp synchronization