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
