cloud-fan commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r4056996899
##########
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:
**Nit (P3):** `respectReliablyStored = true` is also used by every changed
`FetchFailed` cleanup, so describing it as executor or worker loss “rather than
a fetch failure” is already false. The scheduler helper also labels the
host-decommission bypass as merged-chunk unconditional removal, and its
pipelined caller runs before any named-map unregister. Please describe the
flags by their actual semantics and producers consistently across the tracker,
helper, and test comments.
##########
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:
**Non-blocking (P2):** `FetchFailed` removes only the named map before this
bulk pass, and `respectReliablyStored = true` skips every other status in the
same reliable shuffle. If the backend lost several blocks for this executor or
host, retries discover them one at a time and can exhaust
`spark.stage.maxConsecutiveAttempts` instead of recomputing the affected maps
together. Please make this cleanup failure-scoped so correlated target statuses
of the failed shuffle are invalidated while unrelated reliable shuffles remain
preserved.
See **Shared repair plan 1** in the review body.
##########
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:
**Non-blocking (P2):** This skips the entire `ShuffleStatus` when its
original map output is reliable, but push-based registration can also store
host-local `MergeStatus` entries in that same object, and no guard prevents the
two modes from coexisting. After a merger-host failure, only the directly
failed reduce is unregistered; other stale merged chunks on that host remain
advertised and retries can fail again instead of falling back to the reliable
map blocks. Please preserve reliable map statuses separately while still
removing matching merge results.
See **Shared repair plan 1** in the review body.
##########
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:
**Nit (P3):** Please capture the epoch before this mixed cleanup and assert
that it advances. The output-count checks still pass if `shouldBumpEpoch`
incorrectly suppresses the `metadataChanged = true, preservedReliable = true`
case, while the later epoch assertions cover only a reliable-only no-op and
unconditional removal.
--
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]
