SteNicholas commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r4385803431
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4685,27 +4675,53 @@ 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
+ // ignoreShuffleFileLostEpoch (set from host-decommission state)
bypasses the fence, but
+ // reliability filtering still applies. A FetchFailed for a
reliably-stored shuffle is exempt
+ // once per epoch even when the executor fence rejects it: the executor
loss preserved that
+ // shuffle, so its cleanup must still run when it is proven gone.
+ val passesExecutorFence = ignoreShuffleFileLostEpoch ||
+ !shuffleFileLostEpoch.contains(execId) ||
+ shuffleFileLostEpoch(execId) < currentEpoch
+ val reliableFetchFailedBypass = !passesExecutorFence &&
failedShuffleId.exists { id =>
+ mapOutputTracker.isReliablyStored(id) &&
+ reliableShuffleFileLostEpoch.get((execId, id)).forall(_ <
currentEpoch)
}
- if (remove) {
+ if (passesExecutorFence || reliableFetchFailedBypass) {
+ // A bypass-only admission touches just the failed shuffle: only it is
proven gone, so
+ // co-located shuffles (their freshly recomputed maps) must not be
swept.
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,
+ restrictToFailedShuffle = reliableFetchFailedBypass)
+ }
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,
+ restrictToFailedShuffle = reliableFetchFailedBypass)
+ }
+ }
+ // A genuine executor-wide cleanup advances the executor fence
(monotonically) to dedup
+ // same-epoch duplicates. A bypass-only admission must not move it (it
did not clean the
+ // executor); it only records the per-shuffle fence so that shuffle's
own duplicates are
+ // deduped. Both stamps are monotonic. Match prior behavior: skip
under ignore.
+ if (!ignoreShuffleFileLostEpoch) {
+ if (passesExecutorFence) {
+ shuffleFileLostEpoch(execId) = currentEpoch
+ }
+ failedShuffleId.foreach { id =>
+ val key = (execId, id)
+ val prev = reliableShuffleFileLostEpoch.getOrElse(key,
Long.MinValue)
+ reliableShuffleFileLostEpoch(key) = math.max(prev, currentEpoch)
Review Comment:
[P2] Avoid recreating loss fences after terminal job cleanup
When a map-output FetchFailed exhausts the stage retry limit,
handleTaskCompletion calls abortStage before reaching
unregisterOutputsOnFetchFailedExecutor. abortStage synchronously tears down the
job's unshared stages, and cleanupStagesForJob removes their
reliableShuffleFileLostEpoch entries. Execution then continues into this helper
and recreates the (executor, shuffleId) entry here, including for ordinary
local shuffles. The map stage is already gone, so its cleanup cannot retire the
new entry; unless the same executor ID is added again or the same shuffle is
reused and later cleaned up, the entry remains for the application's lifetime.
Repeated terminal failures with distinct shuffles therefore retain unbounded
driver metadata.
Please record the fence before teardown, or avoid recording it for a shuffle
whose map stage has already been removed. Add a regression that sets
spark.stage.maxConsecutiveAttempts to 1, delivers a map-output FetchFailed, and
asserts reliableShuffleFileLostEpochSize is zero after job abortion. The
existing retirement test cancels the job after a non-terminal failure, so it
does not cover this ordering.
--
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]
