[jira] [Created] (FLINK-40517) Checkpointing-during-recovery drops cross-channel watermark / watermark-status aggregation on fan-in rescale

["Rui Fan (Jira)" <ji...@apache.org>]

Rui Fan created FLINK-40517:
-------------------------------

             Summary: Checkpointing-during-recovery drops cross-channel watermark / watermark-status aggregation on fan-in rescale
                 Key: FLINK-40517
                 URL: https://issues.apache.org/jira/browse/FLINK-40517
             Project: Flink
          Issue Type: Sub-task
          Components: Runtime / Checkpointing
            Reporter: Rui Fan
            Assignee: Rui Fan


On the normal recovery path, when a rescale fans several old input channels into one new channel, [`DemultiplexingRecordDeserializer.getNextRecord`](https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/recovery/DemultiplexingRecordDeserializer.java#L108-L131) aggregates the recovered non-record elements across the merged old channels: it emits the **min** watermark (suppressing it until every merged channel has one) and `ACTIVE` if **any** channel is active. This reproduces the pre-rescale `min(wmA, wmB, ...)` semantics so the operator watermark is never advanced past what any old channel justifies.


With *checkpointing during recovery* (CDR, `execution.checkpointing.during-recovery.enabled`) enabled, this aggregation moves to the unspilling thread, but [`GateFilterHandler.filterAndRewrite`](https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java#L294-L302) only filters records and passes watermarks and statuses through **verbatim**. So on a fan-in rescale a high watermark from one merged channel advances the operator watermark prematurely and legitimately in-flight lower-timestamp records get dropped as late (silent data loss), and an `IDLE` from one merged channel wrongly marks the whole channel idle (broken alignment; no data loss).


**Trigger**: CDR enabled + unaligned-checkpoint restore with in-flight state + a fan-in rescale (scale-down / ambiguous keyed exchange); the data-loss symptom additionally needs event-time watermarks and a late-dropping operator. Without CDR the same rescale is correct.


**Fix**: do the same min-watermark / any-active aggregation in the unspilling thread, where the per-old-channel `VirtualChannel`s already hold `lastWatermark` / `watermarkStatus`, instead of passing them through verbatim (`NO_RESCALE` needs no change). Since segments are written per new channel in arrival order, a watermark must be held until all old channels folding into that new channel are drained.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)