Hi Roman, Krishna,
I've created a ticket to track this with a test case to reproduce the
issue [1].
Thank you very much for the response and the idea about borrowed
watermarks. Agreed on both of your points: a group where every reader
is idle needs its own rule, and readers would have to stop sending IDLE
altogether. I found a few other issues though:
1. If the borrowed watermarks are shared in the whole job, it's
possible that completely unrelated sources can either push or stall
other sources.
a. Source A only receives a batch of records every 15 minutes,
Source B is busy, near real time, Source A goes idle, then
Source B's watermark will push Source A's watermark forward,
eventually dropping the batch of records as late.
b. Source A is 7 days behind, it will hold back all other idle
sources, so a join of a busy and idle source can stall.
2. We could re-use the idea of having alignment groups, which could fix
the issues where the sources are unrelated. But it still has a few
shortcomings:
a. Source A is idle, Source B is busy, both are in a group. In one
operator, A and B are joined, which works today. Another operator
is a window over A, but A will now also carry B's watermarks, so
late data might be dropped here.
b. Sources outside any group will keep the race. Users would need to
know about the race itself, know which sources to group, and be
aware of the consequences. Also jobs topology might change, which
will require users to re-think these groups.
Overall I prefer the resume marker proposal I've drawn up in the
original email, even though it also has a few drawbacks:
1. It changes the wire format, so resuming from unaligned checkpoints
with the markers enabled would fail on older Flink versions.
Savepoints should not be affected. Markers skip the operator chain,
so operator interfaces don't change.
2. On resume, the operators will wait for the slowest path. Should be
milliseconds, and we could add a configuration which defines how
long to wait for maximum.
Please let me know what you think and whether you agree with the resume
markers. It doesn't change what idleness means for existing jobs, and a
few versions after users have experimented with it, we could try to
enable it by default, and the correctness would be fixed without the
users needing to know about the race.
Best Regards,
Mate
[1] https://issues.apache.org/jira/browse/FLINK-41268
On Thu, Oct 29, 2026 at 4:50 PM Krishna Sai <[email protected]> wrote:
> 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
>> >
>>
>
