venkata91 commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r4067612223
##########
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:
Fixed in the latest commit. `removeExecutorAndUnregisterOutputs` now stamps
`shuffleFileLostEpoch(execId)` on every real cleanup, so a duplicate same-epoch
FetchFailed for the already-cleaned local shuffle is fenced instead of
bulk-deleting its recomputed maps. The first FetchFailed for a preserved
reliable shuffle is still allowed through a narrow per-`(executor, shuffleId)`
fence (`reliableShuffleFileLostEpoch`), cleared in `handleExecutorAdded`. Added
the requested regression: two same-epoch FetchFailures for a local shuffle
while a co-located reliable shuffle is preserved, asserting the executor
cleanup runs once.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4627,7 +4632,11 @@ private[spark] class DAGScheduler(
// proceed with unconditional removal of shuffle outputs from all
executors on that
// host, including from those that we still haven't confirmed as lost
due to heartbeat
// delays.
- ignoreShuffleFileLostEpoch = isHostDecommissioned)
+ ignoreShuffleFileLostEpoch = isHostDecommissioned,
+ // This FetchFailed only proves the specifically-failed map output is
gone (already
+ // unregistered by name above); an unrelated reliable shuffle on the
same executor lives
+ // off-executor and survives, so preserve reliably-stored shuffles
here too.
+ respectReliablyStored = true)
Review Comment:
Done in `13184e19c26`. A map-output FetchFailed now passes `failedShuffleId
= Some(shuffleId)`, which exempts that shuffle from reliable preservation so
all its correlated maps on the executor/host clear in one pass, while unrelated
reliable shuffles stay preserved. Covered by the "clears the failed shuffle's
co-located maps in one pass" regression in DAGSchedulerSuite.
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1043,21 +1096,66 @@ 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 shuffle outputs associated with this host, returning the
aggregate cleanup outcome.
+ *
+ * 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, respectReliablyStored: Boolean):
CleanupOutcome = {
+ val outcome =
removeSelectively(respectReliablyStored)(_.removeOutputsOnHost(host))
+ if (outcome.shouldBumpEpoch) incrementEpoch()
+ outcome
}
/**
- * 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 two-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 map outputs associated with this executor, returning the
aggregate cleanup outcome.
+ *
+ * When `respectReliablyStored` is true (executor loss rather than a fetch
failure), shuffles
+ * whose output is reliably stored off-executor are left intact: losing the
executor does not lose
+ * their output, so unregistering would force a needless map-stage recompute.
+ */
+ def removeOutputsOnExecutor(execId: String, respectReliablyStored: Boolean):
CleanupOutcome = {
+ val outcome =
removeSelectively(respectReliablyStored)(_.removeOutputsOnExecutor(execId))
+ if (outcome.shouldBumpEpoch) incrementEpoch()
+ outcome
+ }
+
+ /**
+ * Applies `remove` to each shuffle, skipping reliably-stored shuffles when
+ * `respectReliablyStored` is set. Aggregates the outcome: `metadataChanged`
if any removal
+ * actually changed state, and `preservedReliable` if any (one or more)
reliable shuffle was
+ * skipped. Does not bump the epoch; see `CleanupOutcome.shouldBumpEpoch`.
+ */
+ private def removeSelectively(respectReliablyStored: Boolean)(
+ remove: ShuffleStatus => Boolean): CleanupOutcome = {
+ var metadataChanged = false
+ var preservedReliable = false
+ shuffleStatuses.valuesIterator.foreach { status =>
+ if (respectReliablyStored && status.isReliablyStored) {
Review Comment:
Done. `removeOutputsOnHost` now always removes matching `MergeStatus`
entries via `removeMergeResultsByFilter(filter)` regardless of reliability, and
only the reliable map statuses are preserved (`removeMapOutputsSelectively`).
So a merger-host failure clears stale merged chunks on that host while reliable
map blocks remain available for fallback.
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1043,21 +1096,66 @@ 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 shuffle outputs associated with this host, returning the
aggregate cleanup outcome.
+ *
+ * When `respectReliablyStored` is true (executor/worker loss rather than a
fetch failure),
Review Comment:
Done. Reworded the tracker, scheduler-helper, and test comments to describe
`respectReliablyStored` and the `ignoreShuffleFileLostEpoch` bypass by their
actual semantics and producers, rather than framing them as
executor/worker-loss vs fetch-failure.
##########
core/src/test/scala/org/apache/spark/MapOutputTrackerSuite.scala:
##########
@@ -136,6 +136,46 @@ class MapOutputTrackerSuite extends SparkFunSuite with
LocalSparkContext {
rpcEnv.shutdown()
}
+ test("SPARK-59138: executor loss skips reliably-stored shuffles but not
local-disk ones") {
+ val rpcEnv = createRpcEnv("test")
+ val tracker = newTrackerMaster()
+ tracker.trackerEndpoint =
rpcEnv.setupEndpoint(MapOutputTracker.ENDPOINT_NAME,
+ new MapOutputTrackerMasterEndpoint(rpcEnv, tracker, conf))
+
+ val size = MapStatus.compressSize(1000L)
+ // Shuffle 0: local-disk (not reliably stored). Shuffle 1: reliably stored
off-executor.
+ tracker.registerShuffle(0, 1, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES)
+ tracker.registerShuffle(1, 1, MergeStatus.SHUFFLE_PUSH_DUMMY_NUM_REDUCES,
+ isReliablyStored = true)
+ tracker.registerMapOutput(0, 0, MapStatus(BlockManagerId("a", "hostA",
1000), Array(size), 5))
+ tracker.registerMapOutput(1, 0, MapStatus(BlockManagerId("a", "hostA",
1000), Array(size), 6))
+
+ assert(tracker.isReliablyStored(0) === false)
+ assert(tracker.isReliablyStored(1) === true)
+
+ // Executor loss: skip reliably-stored shuffles. Shuffle 0 drops, shuffle
1 stays.
Review Comment:
Done. Added the mixed-cleanup case in MapOutputTrackerSuite that captures
the epoch before and asserts it advances (`metadataChanged = true,
preservedReliable = true`), so a regression in `shouldBumpEpoch` for that case
would be caught.
--
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]
