cloud-fan commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r4085168311
##########
core/src/main/scala/org/apache/spark/shuffle/ShuffleHandle.scala:
##########
@@ -25,4 +25,15 @@ import org.apache.spark.annotation.DeveloperApi
* @param shuffleId ID of the shuffle
*/
@DeveloperApi
-abstract class ShuffleHandle(val shuffleId: Int) extends Serializable {}
+abstract class ShuffleHandle(val shuffleId: Int) extends Serializable {
+ /**
+ * Whether this shuffle's output is stored reliably, off the executor that
produced it (e.g. a
+ * remote shuffle service or distributed filesystem). When true, losing the
executor or its host
+ * does not lose the output, so its map outputs are not unregistered on
executor/worker loss.
+ *
+ * Per-shuffle override of the app-global
`ShuffleDriverComponents.supportsReliableStorage()`:
Review Comment:
**Non-blocking (P2):** `Some(false)` makes this shuffle explicitly
executor-local even when `supportsReliableStorage()` is true, but executor-hold
support and dynamic-allocation admission still trust only the global flag. In a
mixed/fallback application they can therefore retire all executors while an
active local shuffle has no durable copy, violating the hold guarantee and
forcing recomputation. The per-shuffle registry needs to participate in those
application-wide eligibility checks, or this override state needs to be
excluded.
**Recommended change:** Expose a thread-safe aggregate of active
ordinary-shuffle reliability from the tracker and require hold and
dynamic-allocation scale-down decisions to refuse retirement whenever any
active shuffle is explicitly unreliable.
**Why this works:** Keep the global capability as the None fallback and
configuration bootstrap, but make runtime retirement conditional on the
resolved active per-shuffle registry. Re-evaluate the aggregate at each hold or
executor-removal decision so registering or unregistering a fallback shuffle
changes admission without duplicating reliability resolution in consumers.
**Scope:** Connect registered per-shuffle reliability to application-wide
executor retirement and add mixed-mode lifecycle coverage.
**Compatibility:** Per-shuffle Some(true)/Some(false) remains authoritative
in DAGScheduler and TaskSetManager, while None keeps the legacy global fallback.
**Risks:** A stale or incorrectly synchronized aggregate could allow unsafe
retirement or permanently suppress scale-down. An empty active registry must
not be confused with a manager that can never produce unreliable shuffles.
**Constraints:** Some(false) must remain authoritative over global true.
Legacy None plus global true behavior must remain available when all active
shuffles resolve reliable. The lifecycle query must track shuffle registration
and unregistration safely across scheduler and allocation threads.
**Success:** Hold and ordinary dynamic-allocation retirement cannot evict
executors while an active ordinary shuffle resolves unreliable and no
independent preservation mechanism protects it. Removing the last active
unreliable shuffle restores the all-reliable admission path. Per-shuffle
cleanup and task resubmission continue to use the same resolved Boolean.
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1043,21 +1126,76 @@ 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).
+ * outputs which are served by an external shuffle server (if one exists).
Unconditional cleanup:
+ * every shuffle's output on the host is dropped. See the two-argument
overload for the selective
+ * (reliable-storage-preserving) variant.
*/
- def removeOutputsOnHost(host: String): Unit = {
- shuffleStatuses.valuesIterator.foreach { _.removeOutputsOnHost(host) }
- incrementEpoch()
- }
+ def removeOutputsOnHost(host: String): Unit =
+ removeOutputsOnHost(host, respectReliablyStored = false)
/**
- * Removes all shuffle outputs associated with this executor. Note that this
will also remove
+ * Removes all map outputs associated with this executor. Note that this
will also remove
* outputs which are served by an external shuffle server (if one exists),
as they are still
- * registered with this execId.
+ * registered with this execId. Unconditional cleanup: every shuffle's
output on the executor is
+ * dropped. See the three-argument overload for the selective variant.
*/
- def removeOutputsOnExecutor(execId: String): Unit = {
- shuffleStatuses.valuesIterator.foreach { _.removeOutputsOnExecutor(execId)
}
- incrementEpoch()
+ def removeOutputsOnExecutor(execId: String): Unit =
+ removeOutputsOnExecutor(execId, respectReliablyStored = false)
+
+ /**
+ * Removes shuffle outputs on this host. With `respectReliablyStored` true,
reliably-stored
+ * shuffles are kept, except `failedShuffleId` (the shuffle whose fetch
failed), which is cleared
Review Comment:
**Nit (P3):** This Scaladoc promises that `failedShuffleId` is exempted from
preservation, but this overload always delegates with `None`; only the
following three-argument overload can supply that value. Please put the
failed-shuffle semantics on the overload that accepts the parameter so callers
are not promised behavior this signature cannot provide.
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -929,23 +994,41 @@ private[spark] class MapOutputTrackerMaster(
shuffleStatuses.valuesIterator.count(_.hasCachedSerializedBroadcast)
}
- def registerShuffle(shuffleId: Int, numMaps: Int, numReduces: Int): Unit = {
+ def registerShuffle(shuffleId: Int, numMaps: Int, numReduces: Int): Unit =
+ registerShuffle(shuffleId, numMaps, numReduces, isReliablyStored = false)
+
+ def registerShuffle(
+ shuffleId: Int,
+ numMaps: Int,
+ numReduces: Int,
+ isReliablyStored: Boolean): Unit = {
+ // isReliablyStored is orthogonal to push-based shuffle; the branches
differ only in tracking
+ // merge status (numReduces). pushBasedShuffleEnabled is app-global while
reliability is
+ // per-shuffle, so no shuffle is both: a reliable manager (e.g. Celeborn)
intercepts it, and its
Review Comment:
**Nit (P3):** This says no shuffle can be both push-based and reliably
stored, but the branch below stores reliability when push-based tracking is
enabled and the new host-loss test registers exactly that combination. Since
the map and merge cleanup split relies on this supported state, please describe
the actual orthogonality instead of claiming mutual exclusion.
##########
core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala:
##########
@@ -1112,11 +1124,184 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
completeShuffleMapStageSuccessfully(0, 0, 1)
runEvent(ExecutorLost("hostA-exec", event))
verify(blockManagerMaster, times(1)).removeExecutorAsync("hostA-exec")
- verify(mapOutputTracker, times(0)).removeOutputsOnExecutor("hostA-exec")
+ verify(mapOutputTracker, times(0)).removeOutputsOnExecutor("hostA-exec",
true)
assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId,
0).map(_._1).toSet ===
HashSet(makeBlockManagerId("hostA"), makeBlockManagerId("hostB")))
}
+ test("SPARK-59138: executor loss keeps a reliably-stored shuffle but drops a
local-disk one") {
+ // No external shuffle service, so a plain (local-disk) shuffle's outputs
are lost on executor
+ // loss, but a per-shuffle reliably-stored one survives.
+ conf.set(config.SHUFFLE_SERVICE_ENABLED.key, "false")
+
+ val reliableRdd = new MyRDD(sc, 2, Nil)
+ val reliableDep = new ReliablyStoredShuffleDependency(reliableRdd, new
HashPartitioner(1))
+ val localRdd = new MyRDD(sc, 2, Nil)
+ val localDep = new ShuffleDependency(localRdd, new HashPartitioner(1))
+ val reduceRdd = new MyRDD(sc, 1, List(reliableDep, localDep), tracker =
mapOutputTracker)
+ submit(reduceRdd, Array(0))
+
+ // Two independent map stages, one per shuffle; complete both on hostA /
hostB.
+ completeShuffleMapStageSuccessfully(0, 0, 1)
+ completeShuffleMapStageSuccessfully(1, 0, 1)
+
+ runEvent(ExecutorLost("hostA-exec", ExecutorKilled))
+ verify(mapOutputTracker, times(1)).removeOutputsOnExecutor("hostA-exec",
true)
+
+ // Reliable shuffle keeps hostA's output; local-disk shuffle loses it.
+ assert(mapOutputTracker.getMapSizesByExecutorId(reliableDep.shuffleId,
0).map(_._1).toSet ===
+ HashSet(makeBlockManagerId("hostA"), makeBlockManagerId("hostB")))
+ intercept[MetadataFetchFailedException] {
+ mapOutputTracker.getMapSizesByExecutorId(localDep.shuffleId, 0)
+ }
+ }
+
+ test("SPARK-59138: per-shuffle reliablyStored=false overrides a global
supportsReliableStorage") {
+ // Global flag true, but one shuffle reports Some(false); the per-shuffle
value must win.
+ conf.set(config.SHUFFLE_SERVICE_ENABLED.key, "false")
+ conf.set(config.SHUFFLE_IO_PLUGIN_CLASS.key,
+ classOf[TestShuffleDataIOWithMockedComponents].getName)
+ when(sc.shuffleDriverComponents.supportsReliableStorage()).thenReturn(true)
+
+ val reliableRdd = new MyRDD(sc, 2, Nil)
+ val reliableDep = new ReliablyStoredShuffleDependency(reliableRdd, new
HashPartitioner(1))
+ val fallbackRdd = new MyRDD(sc, 2, Nil)
+ val fallbackDep =
+ new ReliablyStoredShuffleDependency(fallbackRdd, new HashPartitioner(1),
Some(false))
+ val reduceRdd = new MyRDD(sc, 1, List(reliableDep, fallbackDep), tracker =
mapOutputTracker)
+ submit(reduceRdd, Array(0))
+
+ completeShuffleMapStageSuccessfully(0, 0, 1)
+ completeShuffleMapStageSuccessfully(1, 0, 1)
+
+ runEvent(ExecutorLost("hostA-exec", ExecutorKilled))
+
+ // Some(true) keeps hostA's output despite the global flag; Some(false)
loses it despite it.
+ assert(mapOutputTracker.getMapSizesByExecutorId(reliableDep.shuffleId,
0).map(_._1).toSet ===
+ HashSet(makeBlockManagerId("hostA"), makeBlockManagerId("hostB")))
+ intercept[MetadataFetchFailedException] {
+ mapOutputTracker.getMapSizesByExecutorId(fallbackDep.shuffleId, 0)
+ }
+ }
+
+ test("SPARK-59138: same-epoch FetchFailed after selective executor-loss
cleanup is not skipped") {
+ // A selective cleanup preserves reliable outputs, so it must not stamp
shuffleFileLostEpoch;
Review Comment:
**Nit (P3):** These comments say selective cleanup leaves
`shuffleFileLostEpoch` unstamped and that the same-epoch failure performs
another selective removal. The current code stamps the executor fence after an
admitted cleanup; the reliable case proceeds through the per-shuffle bypass,
while the local duplicate is skipped. The assertions can pass while teaching
the obsolete state model, so please update the narrative to name the two-level
fence actually under test.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4043,7 +4057,10 @@ private[spark] class DAGScheduler(
}
// TODO: mark the executor as failed only if there were lots of
fetch failures on it
- unregisterOutputsOnFetchFailedExecutor(bmAddress, task)
+ // Exempt the failed shuffle from reliable-preservation only on a
map-output failure
+ // (mapIndex != -1); a merged-chunk failure says nothing about its
reliable map output.
+ val failedMapShuffleId = if (mapIndex != -1) Some(shuffleId) else
None
Review Comment:
**Non-blocking (P2):** The `mapIndex == -1` branch must preserve reliable
original map output while handling the missing merged chunk, but current tests
separate reliable map failures from ordinary merged-chunk failures. Please add
a DAGScheduler case combining a reliably stored shuffle with a merged-chunk
failure so changing this `None` path to `Some(shuffleId)` fails instead of
silently discarding valid fallback maps.
##########
core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala:
##########
@@ -1112,11 +1124,184 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
completeShuffleMapStageSuccessfully(0, 0, 1)
runEvent(ExecutorLost("hostA-exec", event))
verify(blockManagerMaster, times(1)).removeExecutorAsync("hostA-exec")
- verify(mapOutputTracker, times(0)).removeOutputsOnExecutor("hostA-exec")
+ verify(mapOutputTracker, times(0)).removeOutputsOnExecutor("hostA-exec",
true)
assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId,
0).map(_._1).toSet ===
HashSet(makeBlockManagerId("hostA"), makeBlockManagerId("hostB")))
}
+ test("SPARK-59138: executor loss keeps a reliably-stored shuffle but drops a
local-disk one") {
+ // No external shuffle service, so a plain (local-disk) shuffle's outputs
are lost on executor
+ // loss, but a per-shuffle reliably-stored one survives.
+ conf.set(config.SHUFFLE_SERVICE_ENABLED.key, "false")
+
+ val reliableRdd = new MyRDD(sc, 2, Nil)
+ val reliableDep = new ReliablyStoredShuffleDependency(reliableRdd, new
HashPartitioner(1))
+ val localRdd = new MyRDD(sc, 2, Nil)
+ val localDep = new ShuffleDependency(localRdd, new HashPartitioner(1))
+ val reduceRdd = new MyRDD(sc, 1, List(reliableDep, localDep), tracker =
mapOutputTracker)
+ submit(reduceRdd, Array(0))
+
+ // Two independent map stages, one per shuffle; complete both on hostA /
hostB.
+ completeShuffleMapStageSuccessfully(0, 0, 1)
+ completeShuffleMapStageSuccessfully(1, 0, 1)
+
+ runEvent(ExecutorLost("hostA-exec", ExecutorKilled))
+ verify(mapOutputTracker, times(1)).removeOutputsOnExecutor("hostA-exec",
true)
+
+ // Reliable shuffle keeps hostA's output; local-disk shuffle loses it.
+ assert(mapOutputTracker.getMapSizesByExecutorId(reliableDep.shuffleId,
0).map(_._1).toSet ===
+ HashSet(makeBlockManagerId("hostA"), makeBlockManagerId("hostB")))
+ intercept[MetadataFetchFailedException] {
+ mapOutputTracker.getMapSizesByExecutorId(localDep.shuffleId, 0)
+ }
+ }
+
+ test("SPARK-59138: per-shuffle reliablyStored=false overrides a global
supportsReliableStorage") {
+ // Global flag true, but one shuffle reports Some(false); the per-shuffle
value must win.
+ conf.set(config.SHUFFLE_SERVICE_ENABLED.key, "false")
+ conf.set(config.SHUFFLE_IO_PLUGIN_CLASS.key,
+ classOf[TestShuffleDataIOWithMockedComponents].getName)
+ when(sc.shuffleDriverComponents.supportsReliableStorage()).thenReturn(true)
+
+ val reliableRdd = new MyRDD(sc, 2, Nil)
+ val reliableDep = new ReliablyStoredShuffleDependency(reliableRdd, new
HashPartitioner(1))
+ val fallbackRdd = new MyRDD(sc, 2, Nil)
+ val fallbackDep =
+ new ReliablyStoredShuffleDependency(fallbackRdd, new HashPartitioner(1),
Some(false))
+ val reduceRdd = new MyRDD(sc, 1, List(reliableDep, fallbackDep), tracker =
mapOutputTracker)
+ submit(reduceRdd, Array(0))
+
+ completeShuffleMapStageSuccessfully(0, 0, 1)
+ completeShuffleMapStageSuccessfully(1, 0, 1)
+
+ runEvent(ExecutorLost("hostA-exec", ExecutorKilled))
+
+ // Some(true) keeps hostA's output despite the global flag; Some(false)
loses it despite it.
+ assert(mapOutputTracker.getMapSizesByExecutorId(reliableDep.shuffleId,
0).map(_._1).toSet ===
+ HashSet(makeBlockManagerId("hostA"), makeBlockManagerId("hostB")))
+ intercept[MetadataFetchFailedException] {
+ mapOutputTracker.getMapSizesByExecutorId(fallbackDep.shuffleId, 0)
+ }
+ }
+
+ test("SPARK-59138: same-epoch FetchFailed after selective executor-loss
cleanup is not skipped") {
+ // A selective cleanup preserves reliable outputs, so it must not stamp
shuffleFileLostEpoch;
+ // a same-epoch FetchFailed for a preserved-but-gone output must still
trigger real removal.
+ conf.set(config.SHUFFLE_SERVICE_ENABLED.key, "false")
+
+ val shuffleMapRdd = new MyRDD(sc, 2, Nil)
+ val shuffleDep = new ReliablyStoredShuffleDependency(shuffleMapRdd, new
HashPartitioner(2))
+ val shuffleId = shuffleDep.shuffleId
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker =
mapOutputTracker)
+ submit(reduceRdd, Array(0, 1))
+ completeShuffleMapStageSuccessfully(0, 0, reduceRdd.partitions.length)
+
+ // Executor loss: reliably-stored shuffle is preserved (selective
cleanup), no output removed.
+ runEvent(ExecutorLost("hostA-exec", ExecutorKilled))
+ assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId,
0).map(_._1.host).toSet ===
+ HashSet("hostA", "hostB"))
+
+ // Same-epoch FetchFailed for hostA's (really gone) output: no epoch was
stamped, so the
+ // epoch-gated cleanup proceeds instead of being skipped. The executor
loss preserved the
+ // reliable shuffle (2-arg cleanup); this map-output FetchFailed exempts
the failed shuffle.
+ complete(taskSets(1), Seq(
+ (Success, 42),
+ (FetchFailed(makeBlockManagerId("hostA"), shuffleId, 0L, 0, 1,
"ignored"), null)))
+ verify(mapOutputTracker, times(1)).removeOutputsOnExecutor("hostA-exec",
true)
+ verify(mapOutputTracker, times(1)).removeOutputsOnExecutor("hostA-exec",
true, Some(shuffleId))
+ }
+
+ test("SPARK-59138: same-epoch FetchFailed for a local shuffle preserves a
reliable one") {
+ // Executor loss selectively drops the local-disk shuffle but keeps the
reliable one. A
+ // same-epoch FetchFailed for the local shuffle must not bulk-drop the
reliable shuffle that
+ // shares the executor, since its output lives off-executor.
+ conf.set(config.SHUFFLE_SERVICE_ENABLED.key, "false")
+
+ val reliableRdd = new MyRDD(sc, 2, Nil)
+ val reliableDep = new ReliablyStoredShuffleDependency(reliableRdd, new
HashPartitioner(1))
+ val localRdd = new MyRDD(sc, 2, Nil)
+ val localDep = new ShuffleDependency(localRdd, new HashPartitioner(1))
+ val reduceRdd = new MyRDD(sc, 1, List(reliableDep, localDep), tracker =
mapOutputTracker)
+ submit(reduceRdd, Array(0))
+
+ completeShuffleMapStageSuccessfully(0, 0, 1)
+ completeShuffleMapStageSuccessfully(1, 0, 1)
+
+ runEvent(ExecutorLost("hostA-exec", ExecutorKilled))
+
+ // Same-epoch FetchFailed for the local shuffle. It removes non-reliable
output only, so the
+ // reliable shuffle keeps hostA's output.
+ complete(taskSets(2), Seq(
+ (FetchFailed(makeBlockManagerId("hostA"), localDep.shuffleId, 0L, 0, 0,
"ignored"), null)))
+ assert(mapOutputTracker.getMapSizesByExecutorId(reliableDep.shuffleId,
0).map(_._1).toSet ===
+ HashSet(makeBlockManagerId("hostA"), makeBlockManagerId("hostB")))
+ }
+
+ test("SPARK-59138: map-output FetchFailed clears the failed shuffle's
co-located maps in one " +
+ "pass while an unrelated reliable shuffle on the same executor survives") {
+ // A FetchFailed against a reliably-stored map output is evidence that
shuffle's output on the
+ // host is actually gone, so the failed shuffle is exempted from reliable
preservation and all
+ // its maps on hostA clear together. A different reliable shuffle on hostA
has no such evidence
+ // and stays registered.
+ conf.set(config.SHUFFLE_SERVICE_ENABLED.key, "false")
+
+ val failedRdd = new MyRDD(sc, 2, Nil)
+ val failedDep = new ReliablyStoredShuffleDependency(failedRdd, new
HashPartitioner(1))
+ val otherRdd = new MyRDD(sc, 1, Nil)
+ val otherDep = new ReliablyStoredShuffleDependency(otherRdd, new
HashPartitioner(1))
+ val reduceRdd = new MyRDD(sc, 1, List(failedDep, otherDep), tracker =
mapOutputTracker)
+ submit(reduceRdd, Array(0))
+
+ // Put every map of both shuffles on hostA (the 1-task stage uses only the
first entry). The
+ // failed shuffle's two hostA maps must both clear via the FetchFailed
exemption; the unrelated
+ // reliable shuffle's hostA map stays registered.
+ completeShuffleMapStageSuccessfully(0, 0, 1, hostNames = Seq("hostA",
"hostA"))
+ completeShuffleMapStageSuccessfully(1, 0, 1, hostNames = Seq("hostA",
"hostA"))
+
+ complete(taskSets(2), Seq(
+ (FetchFailed(makeBlockManagerId("hostA"), failedDep.shuffleId, 0L, 0, 0,
"ignored"), null)))
+
+ // The failed shuffle loses both hostA maps in one pass (it is exempted
from preservation).
+ // getNumAvailableOutputs proves both are gone: the old per-map behavior
would leave one.
+ assert(mapOutputTracker.getNumAvailableOutputs(failedDep.shuffleId) === 0)
+ intercept[MetadataFetchFailedException] {
+ mapOutputTracker.getMapSizesByExecutorId(failedDep.shuffleId, 0)
+ }
+ // The unrelated reliable shuffle keeps hostA's output.
+ assert(mapOutputTracker.getMapSizesByExecutorId(otherDep.shuffleId,
0).map(_._1).toSet ===
+ HashSet(makeBlockManagerId("hostA")))
+ }
+
+ test("SPARK-59138: duplicate same-epoch FetchFailed for a local shuffle is
fenced even when a " +
Review Comment:
**Non-blocking (P2):** This duplicate regression uses a local shuffle, so it
never exercises `reliableFetchFailedBypass` or `reliableShuffleFileLostEpoch`.
A missing write or read of the new per-shuffle fence would still pass while
repeated reliable-shuffle failures could clear freshly recomputed maps. Please
deliver two same-epoch failures for the same reliably stored shuffle and assert
that only the first reaches cleanup.
See **Shared repair plan 1** in the review body.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4674,27 +4715,42 @@ 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 (merged-chunk failure) removes everything
past the gate. A
+ // FetchFailed for a reliably-stored shuffle is exempt once per epoch:
the executor loss
+ // preserved it, so the executor fence would otherwise skip clearing it
now proven gone.
+ val reliableFetchFailedBypass = failedShuffleId.exists { id =>
+ mapOutputTracker.isReliablyStored(id) &&
+ reliableShuffleFileLostEpoch.get((execId, id)).forall(_ <
currentEpoch)
}
- if (remove) {
+ val shouldRemove = ignoreShuffleFileLostEpoch ||
+ reliableFetchFailedBypass ||
+ !shuffleFileLostEpoch.contains(execId) ||
+ shuffleFileLostEpoch(execId) < currentEpoch
+ if (shouldRemove) {
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)
+ }
+ }
+ // Stamp the executor fence on every real cleanup to dedup same-epoch
duplicates; also stamp
+ // the per-shuffle fence so a bypassed reliable shuffle's own
duplicates are deduped. Match
+ // prior behavior: don't stamp under ignoreShuffleFileLostEpoch.
+ if (!ignoreShuffleFileLostEpoch) {
+ shuffleFileLostEpoch(execId) = currentEpoch
+ failedShuffleId.foreach(id => reliableShuffleFileLostEpoch((execId,
id)) = currentEpoch)
Review Comment:
**Non-blocking (P2):** Every successful map-output `FetchFailed` writes an
`(execId, shuffleId)` entry here, including for non-reliable shuffles. These
entries are removed only if that exact executor ID is added again; shuffle
unregistration and replacement executors with new IDs do not retire them, so
long-running failure-heavy applications retain unbounded scheduler metadata.
See **Shared repair plan 1** in the review body.
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -369,30 +398,60 @@ private class ShuffleStatus(
}
/**
- * 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).
+ * Removes this shuffle's outputs on `host`. Map output is preserved when
`respectReliablyStored`
+ * is set and the shuffle is reliably stored (it lives off the host). Merge
results are always
+ * removed: a `MergeStatus` sits on its merger host, so one matching `host`
is genuinely gone and
+ * an off-host one does not match the filter. Also removes outputs served by
an external shuffle
+ * server, if any.
*/
- def removeOutputsOnHost(host: String): Unit = withWriteLock {
- logDebug(s"Removing outputs for host ${host}")
- removeOutputsByFilter(x => x.host == host)
- removeMergeResultsByFilter(x => x.host == host)
- }
+ def removeOutputsOnHost(host: String, respectReliablyStored: Boolean):
ShuffleRemovalResult =
+ withWriteLock {
+ logDebug(s"Removing outputs for host ${host}")
+ val filter = (x: BlockManagerId) => x.host == host
+ val mergeChanged = removeMergeResultsByFilter(filter)
+ removeMapOutputsSelectively(filter, respectReliablyStored, mergeChanged)
+ }
/**
- * Removes all map outputs associated with the specified executor. Note that
this will also
- * remove outputs which are served by an external shuffle server (if one
exists), as they are
- * still registered with that execId.
+ * Removes this shuffle's map outputs on the executor, preserving them when
+ * `respectReliablyStored` is set and the shuffle is reliably stored.
Executor loss never matches
+ * merge results, whose locations carry the synthetic merger executor id.
Also removes outputs
+ * served by an external shuffle server, if any, as they are still
registered with that execId.
*/
- def removeOutputsOnExecutor(execId: String): Unit = withWriteLock {
- logDebug(s"Removing outputs for execId ${execId}")
- removeOutputsByFilter(x => x.executorId == execId)
+ def removeOutputsOnExecutor(
+ execId: String,
+ respectReliablyStored: Boolean): ShuffleRemovalResult = withWriteLock {
+ logDebug(s"Removing outputs for execId ${execId}")
+ removeMapOutputsSelectively(x => x.executorId == execId,
respectReliablyStored,
+ alreadyChanged = false)
+ }
+
+ /**
+ * Shared map-output removal for the executor/host selective paths.
`mapPreservedReliable` is true
+ * only when a live map output actually matched `f`. `alreadyChanged` folds
in any merge removal
+ * the caller already performed.
+ */
+ private def removeMapOutputsSelectively(
+ f: BlockManagerId => Boolean,
+ respectReliablyStored: Boolean,
+ alreadyChanged: Boolean): ShuffleRemovalResult = withWriteLock {
+ if (respectReliablyStored && isReliablyStored) {
+ val preserved = mapStatuses.exists(s => s != null && f(s.location))
Review Comment:
**Non-blocking (P2):** This target-scoped `exists` check is what preserves
the epoch bump when reliable outputs exist only on other executors, but the
added tests cover only a matching reliable target and an actual mixed removal.
Please add a case with reliable output registered elsewhere and no output on
the lost target, and assert that the ordinary no-op loss still advances the
epoch; otherwise a regression to shuffle-level classification passes.
##########
core/src/main/scala/org/apache/spark/shuffle/ShuffleHandle.scala:
##########
@@ -25,4 +25,15 @@ import org.apache.spark.annotation.DeveloperApi
* @param shuffleId ID of the shuffle
*/
@DeveloperApi
-abstract class ShuffleHandle(val shuffleId: Int) extends Serializable {}
+abstract class ShuffleHandle(val shuffleId: Int) extends Serializable {
+ /**
+ * Whether this shuffle's output is stored reliably, off the executor that
produced it (e.g. a
Review Comment:
**Non-blocking (P2):** This `@DeveloperApi` contract promises that
`Some(true)` means output survives executor or host loss for any
`ShuffleHandle`. The common registration path also forwards the resolved value
for a `PipelinedShuffleDependency`, but `StreamingShuffleOutputTrackerMaster`
discards it and retains its existing group-abort/task-location recovery. That
lets a supported third-party pipelined manager opt into a guarantee it does not
receive; please explicitly exclude and reject that dependency kind, or
implement the contract for it.
**Recommended change:** Scope reliablyStored to materialized
map-output-tracked shuffles and make a non-empty pipelined value fail clearly
instead of being silently accepted and ignored.
**Why this works:** State the blocking/materialized restriction on the
public method, resolve the value only on the ordinary MapOutputTracker
registration route, and reject Some(_) for PipelinedShuffleDependency before
streaming or tracker-free registration. Remove the unused reliability parameter
from the streaming-only contract when possible.
**Scope:** Align the public reliability domain with dependency-kind
registration and pipelined loss semantics.
**Compatibility:** Pipelined group-atomic failure handling and transient
task-location tracking remain unchanged for supported handles.
**Risks:** A guard placed after partial stage registration could leave
inconsistent scheduler state.
**Constraints:** Blocking shuffle Some(true), Some(false), and None
semantics must remain unchanged. Pipelined managers returning the default None
must keep current behavior. Unsupported non-empty pipelined values must fail
before stage or tracker mutation.
**Success:** The public method explicitly names its supported dependency
domain. No pipelined registration silently accepts a reliability value it
cannot honor. Existing pipelined handles using the default None continue to
register normally.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4674,27 +4715,42 @@ 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 (merged-chunk failure) removes everything
past the gate. A
+ // FetchFailed for a reliably-stored shuffle is exempt once per epoch:
the executor loss
+ // preserved it, so the executor fence would otherwise skip clearing it
now proven gone.
+ val reliableFetchFailedBypass = failedShuffleId.exists { id =>
+ mapOutputTracker.isReliablyStored(id) &&
+ reliableShuffleFileLostEpoch.get((execId, id)).forall(_ <
currentEpoch)
Review Comment:
**Non-blocking (P2):** The bypass checks only the per-shuffle fence. If a
local failure at epoch `e2` has already stamped the executor fence, a delayed
reliable failure from `e1 < e2` with no per-shuffle entry is still admitted,
clears current maps, and assigns the executor fence back to `e1`; failures
through `e2` can then run cleanup again. The bypass must respect a newer
executor epoch, and this assignment must never lower that fence.
See **Shared repair plan 1** in the review body.
--
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]
