[
https://issues.apache.org/jira/browse/FLINK-40517?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40517:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Labels: pull-request-available
>
> 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)
