Hi Roman, Mate, Two things on borrowed watermarks.
With alignment on, SourceOperator#emitLatestWatermark reports Watermark.MAX_WATERMARK for an idle subtask, and WatermarkAggregator keeps the smallest watermark of all keys. So the combined value is already the minimum over the active readers. With every reader idle it is Long.MAX_VALUE, which announceCombinedWatermark notes in its overflow catch. FLINK-32496 set it up that way so an idle source stops holding alignment back, where it is harmless. As a stream watermark it is not: readers emitting what they borrowed would put Long.MAX_VALUE on every channel and fire every event-time timer. An all-idle group needs its own rule. Second, the status has to move with the watermark. RecordWriterOutput#emitWatermark and ChainingOutput#emitWatermark both return early while the announced status is IDLE, so a borrowed watermark from a reader that still announces IDLE never leaves the task. FLINK-40499 is the same drop at drain and #29037 emits ACTIVE first. Best, Sepuri Sai Krishna On Thu, Oct 29, 2026 at 5:21 AM Roman Khachatryan <[email protected]> wrote: > Hi Mate, > > Thanks for the detailed write-up. The scenario sounds plausible to me. I'd > also expect it with bounded out-of-orderness, not just a zero-delay > watermark: it only takes a resume burst wider than the delay, with the > records landing on different subtasks. > > Could you share a reproducer, ideally a test that forces the ordering (e.g. > delaying A until X's watermark advances)? It would make the discussion > concrete and could serve as an acceptance test for whichever fix we choose. > Once you have it, could you open a JIRA ticket with it? > > On the solutions: I agree that flushing on ACTIVE only narrows the window. > Status travels in-band, so the flush removes the buffer-timeout gap between > A's and B's flushers, but it doesn't help when A is slow or backpressured. > My main concern with the marker is tracking several sources and splits that > go through the same input channel. After an all-to-all exchange, each > channel into X carries a mix of all source subtasks, and it has a single > watermark: the minimum over everything upstream. So "keep the channel until > it catches up" isn't well defined for one resuming source, because X can't > tell from the channel's watermark whether the resuming source's data has > passed or whether it reflects other sources. With several sources (or > splits) resuming and going idle concurrently, their markers interleave on > the same channels. Every task would then have to track and forward marker > state per channel and per resuming source, with epochs, and re-including > channels too eagerly would stall on sources that are still idle. On top of > that, it adds a new stream element, which touches the operator interfaces > and the in-flight data of unaligned checkpoints. > > Two alternatives that might be worth considering: > > 1. Resume grace in the source. When a source resumes, it emits records > right away but holds its watermark for a delay D before advancing. That > gives r a head start of D through the whole pipeline. It's a small, local > change in the watermark generator (e.g. a resume delay alongside the idle > timeout). It costs +D event-time latency per resume and is still a > heuristic: the total skew between paths has to stay below D. > > 2. Borrowed watermarks (credit to Arvid Heise for this idea). Keep idleness > internal to the source, and never propagate IDLE downstream. Instead, an > idle reader keeps emitting a watermark it gets from the coordinator, e.g. > the minimum over the active readers, job-wide via the CoordinatorStore. > Since every channel always carries a watermark and is always counted, the > in-order argument holds and the race disappears by construction. Timing > then affects only latency. Most of the plumbing already exists for > watermark alignment (ReportedWatermarkEvent and the periodic aggregation in > SourceCoordinator), and no change to the data-stream wire format is needed. > The costs: idle readers lag by roughly the coordinator interval, mid-stream > watermark generators would need a fallback, and operators that observe > WatermarkStatus would see different behaviour, so it would probably start > as an opt-in mode and need a FLIP. As a side benefit, it should also > address FLINK-41254, since an idle subtask's watermark keeps advancing. > > Regarding your third point: +1 to adding a release note for FLINK-39719. > Jobs that hit this race used to get these rows late and will now lose them > silently. Mika, Rui, can you please make sure this makes it into the 2.4.0 > release notes? > > Regards, > Roman > > > On Tue, Oct 27, 2026 at 4:40 PM Mate Czagany <[email protected]> wrote: > > > Hi devs, > > > > I would like to raise a correctness issue that we've run into with a > > sparse source, which often goes idle. This issue only affects pipelines > > where idleness is enabled and parallelism > 1. > > In some cases, where multiple all-to-all exchange (keyBy/rebalance) > > exists in a pipeline, a record that is on time can reach a downstream > > event-time operator after a watermark equal to its own timestamp. > > > > With source subtask (S), a keyBy to two subtasks A and B, and a temporal > > join operator X after a second keyBy, r being the single record: > > > > S -> A (gets record r) -> X > > S -> B (never sees r) -> X > > > > 1. S has been marked idle, A and B has announced IDLE status, thus X > > leaves both of them out of its minimum watermark. > > 2. S reads r with timestamp T. It sends r to A, then ACTIVE status and > > watermark of T to both A and B. > > 3. B has nothing to process, it forwards ACTIVE status and watermark of > > T right away. > > 4. B's watermark might reach X while A's channel is still marked idle, > > X advances watermark, resulting in stalling or dropping the record > > coming from A, as the record's timestamp is equal to the X's > > watermark > > > > Everything arrives in order in each channel, but it's unpredictable > > which message reaches X first. A single exchange can't trigger it, > > because the record and its watermark share one channel. > > Unbounded OVER aggregations drop the record. The event-time temporal > > join delays it until the next watermark up to Flink version 2.3, then > > drops it from 2.4.0 on (FLINK-39719 [1]). Another ticket, FLINK-40504 > > [2] makes this issue a bit less common but doesn't fix the race fully. > > > > 1. I've found an easy way to mitigate the issue by flushing the network > > output when a task announces ACTIVE in RecordWriterOutput [3]. It's > > only a few lines, and since status changes are rare, the cost should > > be minimal. It really does help in the race, but it only narrows the > > window and doesn't help when subtask A is really just slow. > > > > 2. A complete fix would need a FLIP. Operator X can't solve it on its > > own. My idea is that a resuming source could send a small marker > > ahead of its watermark, every task forwards it immediately, and a > > task that receives it keeps the channels idled by that source in its > > minimum watermark until they catch up. This of course would change > > the idleness protocol and wire format, which is why it would need a > > FLIP. > > > > 3. Separately, I think the 2.4.0 release note for FLINK-39719 should > > say that temporal join probe rows at or below the watermark are now > > dropped, since jobs hitting this race will lose rows they used to > > emit late. > > > > What do you think, is it something that anyone else has run into? Would > > it be worth it to have a FLIP and update the idleness logic for this? > > > > Best Regards, > > Mate > > >
