venkata91 commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r3996952718
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1044,20 +1059,48 @@ private[spark] class MapOutputTrackerMaster(
/**
* Removes all shuffle outputs associated with this host. Note that this
will also remove
* outputs which are served by an external shuffle server (if one exists).
+ *
+ * When `respectReliablyStored` is true (executor/worker loss rather than a
fetch failure),
+ * shuffles whose output is reliably stored off-executor are left intact,
since losing the host
+ * does not lose their output.
*/
- def removeOutputsOnHost(host: String): Unit = {
- shuffleStatuses.valuesIterator.foreach { _.removeOutputsOnHost(host) }
- incrementEpoch()
+ def removeOutputsOnHost(host: String): Unit =
+ removeOutputsOnHost(host, respectReliablyStored = false)
+
+ def removeOutputsOnHost(host: String, respectReliablyStored: Boolean): Unit
= {
+ var removedAny = false
+ shuffleStatuses.valuesIterator.foreach { status =>
+ if (!(respectReliablyStored && status.isReliablyStored)) {
+ status.removeOutputsOnHost(host)
+ removedAny = true
Review Comment:
Reworked. The selective cleanup now returns `CleanupOutcome(metadataChanged,
preservedReliable)`, and we bump the epoch on `metadataChanged ||
!preservedReliable`.
One thing to note: we can't gate on `metadataChanged` alone. A no-op
executor loss (executor had no outputs) used to bump the epoch, and that bump
matters for stale-completion fencing: `executorFailureEpoch(execId)` is stamped
unconditionally, and later completions are fenced by `smt.epoch <=
executorFailureEpoch(execId)`. If the tracker epoch doesn't advance, a valid
post-failure relaunch gets wrongly fenced (covered by the "ignore late map task
completions" test). So we skip the bump only when the pass did nothing but
preserve reliable output, which is the wasteful case you pointed at. Everything
else keeps the old bump.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4297,7 +4303,11 @@ private[spark] class DAGScheduler(
true
} else if (!shuffleFileLostEpoch.contains(execId) ||
shuffleFileLostEpoch(execId) < currentEpoch) {
- shuffleFileLostEpoch(execId) = currentEpoch
+ // A selective cleanup keeps reliably-stored outputs, so it isn't a
full cleanup: don't
+ // stamp the epoch, or a same-epoch FetchFailed for a
preserved-but-gone output is skipped.
+ if (!respectReliablyStored) {
Review Comment:
Fixed without adding new per-loss epoch state. Two parts:
- The FetchFailed cleanup now passes `respectReliablyStored = true`, so it
removes non-reliable output only. The failed map is already unregistered by
name via `unregisterMapOutput` before this call, and reliable output survives
executor loss, so a co-located reliable shuffle is no longer dropped.
- `shuffleFileLostEpoch` is stamped only when `!preservedReliable`. A
reliability-preserving cleanup is partial, so we shouldn't suppress a later
same-epoch FetchFailed for an output that was preserved but is actually gone.
Added a regression test (`SPARK-59138: same-epoch FetchFailed for a local
shuffle preserves a reliable one`): mixed reliable + local-disk shuffle on the
same executor, executor loss then a same-epoch FetchFailed for the local
shuffle, asserting the reliable shuffle keeps its output.
--
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]
