tdcmeehan commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r3960453867
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1228,8 +1228,20 @@ private[spark] class TaskSetManager(
// pipelined set and aborts the whole group. Note isZombie already skips a
fully-complete
// producer's set; this guard also covers a PARTIALLY-complete producer
losing an executor on
// decommission.
+
+ // OR, not AND: a shuffle reliably stored off-executor (globally, or just
this one via a remote
+ // shuffle service) keeps its map output when the executor dies. The
per-shuffle bit only ever
+ // adds reliability (defaults to false, set true solely by an opting-in
manager), so a false
+ // there means "no info", not "unreliable".
+ val reliablyStored =
sched.sc.shuffleDriverComponents.supportsReliableStorage() ||
Review Comment:
Make the per-shuffle handle authoritative here and in
`DAGScheduler.handleExecutorLost`. As written, a manager with
`supportsReliableStorage() == true` that returns a false handle for a local
fallback shuffle still keeps that task successful and skips output cleanup, so
the mixed-storage case remains broken.
##########
core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala:
##########
@@ -1112,11 +1123,38 @@ 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")
Review Comment:
Please cover the actual fallback combination: set
`supportsReliableStorage()` to true while one handle reports `isReliablyStored
= false`. The current setup leaves the global capability false, so it cannot
catch the legacy global checks overriding the new per-shuffle value.
--
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]
