sunchao commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r3961027224
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4199,7 +4200,10 @@ private[spark] class DAGScheduler(
execId = execId,
fileLost = fileLost,
hostToUnregisterOutputs = workerHost,
- maybeEpoch = None)
+ maybeEpoch = None,
+ // Executor loss (not a fetch failure): preserve shuffles whose output
is reliably stored
+ // off-executor. Their data survives the executor, so recomputing them
would be wasteful.
+ skipReliablyStored = true)
Review Comment:
[P2] Keep selective executor-loss cleanup from suppressing fetch-failure
cleanup
With the global reliability flag and external shuffle service disabled, this
call preserves reliable outputs but still records `shuffleFileLostEpoch(execId)
= E`. If an already-running reducer then reports `FetchFailed` at epoch E, the
strict epoch check in `removeExecutorAndUnregisterOutputs` rejects the later
bulk cleanup, even though those reliable outputs were never removed.
`unregisterMapOutput` removes only the named map; its epoch increment does not
help because the bulk cleanup uses `task.epoch`.
I reproduced this with two unavailable map outputs from the lost executor:
the first failure recomputes only one partition, then the remaining stale
output causes another stage failure. With
`spark.stage.maxConsecutiveAttempts=2`, the job aborts; the ordinary-shuffle
control recomputes both and succeeds. This was a synthetic scheduler-event test
using the reviewed sources over cached build dependencies, not a remote-shuffle
deployment.
Please distinguish selective executor-loss cleanup from full fetch-failure
cleanup in the epoch bookkeeping, and cover `ExecutorLost` followed by a
same-epoch `FetchFailed` for a reliably stored shuffle.
##########
core/src/main/scala/org/apache/spark/shuffle/ShuffleHandle.scala:
##########
@@ -25,4 +25,12 @@ 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 outside the executors
that produced it (e.g.
+ * a remote shuffle service). When true, losing an executor does not lose
this shuffle's output,
+ * so its map outputs are not unregistered on executor loss. Defaults to
false; a ShuffleManager
Review Comment:
Please make the host-loss requirement explicit in this contract.
`DAGScheduler.handleWorkerRemoved` and the worker-host executor-loss path also
preserve outputs when this flag is true, but the documentation currently
promises only survival of executor loss. Storage in another process on the same
host can meet that wording while disappearing with the worker/host. Require
storage independent of the executor host's lifecycle, matching the existing
`ShuffleDriverComponents.supportsReliableStorage()` contract.
--
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]
