sunchao commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r4066169776
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4674,27 +4705,35 @@ private[spark] class DAGScheduler(
clearCacheLocs()
}
if (fileLost) {
- // When the fetch failure is for a merged shuffle chunk,
ignoreShuffleFileLostEpoch is true
- // and so all the files will be removed.
- val remove = if (ignoreShuffleFileLostEpoch) {
- true
- } else if (!shuffleFileLostEpoch.contains(execId) ||
- shuffleFileLostEpoch(execId) < currentEpoch) {
- shuffleFileLostEpoch(execId) = currentEpoch
- true
- } else {
- false
- }
- if (remove) {
- hostToUnregisterOutputs match {
+ // A merged-shuffle-chunk fetch failure (ignoreShuffleFileLostEpoch)
removes everything
+ // regardless of the epoch gate; otherwise skip a cleanup already done
at this epoch.
+ val shouldRemove = ignoreShuffleFileLostEpoch ||
+ !shuffleFileLostEpoch.contains(execId) ||
+ shuffleFileLostEpoch(execId) < currentEpoch
+ if (shouldRemove) {
+ val outcome = hostToUnregisterOutputs match {
case Some(host) =>
logInfo(log"Shuffle files lost for host: ${MDC(HOST, host)} (epoch
" +
log"${MDC(EPOCH, currentEpoch)}")
- mapOutputTracker.removeOutputsOnHost(host)
+ failedShuffleId match {
+ case None => mapOutputTracker.removeOutputsOnHost(host,
respectReliablyStored)
+ case failed =>
+ mapOutputTracker.removeOutputsOnHost(host,
respectReliablyStored, failed)
+ }
case None =>
- logInfo(log"Shuffle files lost for executor: ${MDC(EXECUTOR_ID,
execId)} " +
- log"(epoch ${MDC(EPOCH, currentEpoch)})")
- mapOutputTracker.removeOutputsOnExecutor(execId)
+ logInfo(log"Shuffle files lost for executor: ${MDC(EXECUTOR_ID,
execId)} " +
+ log"(epoch ${MDC(EPOCH, currentEpoch)})")
+ failedShuffleId match {
+ case None => mapOutputTracker.removeOutputsOnExecutor(execId,
respectReliablyStored)
+ case failed =>
+ mapOutputTracker.removeOutputsOnExecutor(execId,
respectReliablyStored, failed)
+ }
+ }
+ // Record the lost epoch only for a complete cleanup. A cleanup that
preserved reliable
+ // output is partial, so a later same-epoch FetchFailed for a
preserved-but-gone output must
+ // still be processed. Match prior behavior: don't stamp under
ignoreShuffleFileLostEpoch.
+ if (!ignoreShuffleFileLostEpoch && outcome.isCompleteCleanup) {
Review Comment:
[P2] Retain an epoch fence for the outputs already cleaned
If executor X holds local shuffle L and an unrelated reliable shuffle R, the
first FetchFailed for L clears L but preserves R, so this branch records no
cleanup epoch for X. While L's map stage retries, map 1 can finish on
still-live X at a newer epoch while map 0 is still pending. A delayed
FetchFailed for old L/map 0 from another task of the original reducer attempt
then passes shouldRemove and bulk-deletes the newly recomputed map 1. The
stage-attempt guard does not reject that report because the reducer attempt has
not restarted yet; unregisterMapOutput itself is a no-op for the still-missing
map 0. This causes an unnecessary extra map-stage attempt, whereas the previous
same-epoch fence suppressed the duplicate bulk cleanup.
Please retain cleanup-epoch state for the subset of outputs or shuffles
already invalidated, while still allowing the first fetch failure for a
preserved reliable shuffle. Add a regression that completes one retry map
before delivering a second original-attempt failure for a different,
still-missing map.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
