[
https://issues.apache.org/jira/browse/FLINK-40517?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rui Fan updated FLINK-40517:
----------------------------
Description:
On the normal recovery path, when a rescale fans several old input channels
into one new channel, DemultiplexingRecordDeserializer.getNextRecord [1]
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
ChannelStateFilteringHandler.filterAndRewrite [2] 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 VirtualChannels 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.
[1]
https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/recovery/DemultiplexingRecordDeserializer.java#L108-L131
[2]
https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java#L294-L302
was:
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.
> 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
> Priority: Major
>
> On the normal recovery path, when a rescale fans several old input channels
> into one new channel, DemultiplexingRecordDeserializer.getNextRecord [1]
> 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
> ChannelStateFilteringHandler.filterAndRewrite [2] 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 VirtualChannels 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.
> [1]
> https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/recovery/DemultiplexingRecordDeserializer.java#L108-L131
> [2]
> https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java#L294-L302
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
