venkata91 commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r4434070133
##########
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:
@SteNicholas, thanks, I reproduced this with `maxConsecutiveAttempts=1`: the
abort removed the map stage and its fences, and the handler then wrote a new
one. In e982a26 the handler skips the per-shuffle write once the map stage is
gone. New tests cover terminal aborts for both reliable and ordinary shuffles.
--
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]
