cloud-fan commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r3968978562
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1228,8 +1228,13 @@ 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.
+
+ // The tracker's stored value already folds the app-global flag and the
per-shuffle handle
+ // together at registration (see DAGScheduler.createShuffleMapStage), and
a shuffle map task set
+ // always has its shuffle registered before it is submitted. Read it as
the single source.
+ val reliablyStored =
taskSet.shuffleId.exists(sched.mapOutputTracker.isReliablyStored)
val maybeShuffleMapOutputLoss = isShuffleMapTasks && !taskSet.isPipelined
&&
- !sched.sc.shuffleDriverComponents.supportsReliableStorage() &&
+ !reliablyStored &&
Review Comment:
**Non-blocking (P2):** The TaskSetManager coverage protects only the stored
`true` branch. Please add the inverse fallback case here: make the
application-global capability `true`, register this shuffle as explicitly
unreliable, complete a map task on the executor, then lose or decommission it
and assert that the task is invalidated and resubmitted. DAGSchedulerSuite's
tracker-cleanup assertion does not protect TaskSetManager's separate
`successful` and `Resubmitted` bookkeeping.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4297,7 +4303,11 @@ private[spark] class DAGScheduler(
true
} else if (!shuffleFileLostEpoch.contains(execId) ||
shuffleFileLostEpoch(execId) < currentEpoch) {
- shuffleFileLostEpoch(execId) = currentEpoch
+ // A selective cleanup keeps reliably-stored outputs, so it isn't a
full cleanup: don't
+ // stamp the epoch, or a same-epoch FetchFailed for a
preserved-but-gone output is skipped.
+ if (!respectReliablyStored) {
Review Comment:
**Non-blocking (P2):** Consider executor E with reliable shuffle R and local
shuffle L. Executor loss removes L and preserves R, but because this path
deliberately leaves no full-cleanup epoch marker, an in-flight same-epoch
FetchFailed for L passes the gate below. Its named-map unregister is harmless,
then `respectReliablyStored = false` bulk cleanup deletes R even though L's
failure is not evidence that R's remote output disappeared. Please retain
same-epoch selective-cleanup state so the failed map can be invalidated while
later executor/host cleanup continues to preserve unrelated reliable shuffles,
and add this mixed R/L sequence to DAGSchedulerSuite.
**Recommended change:** Record whether an executor or host was selectively
cleaned at the task epoch and keep the later FetchFailed bulk cleanup selective
after unregistering the specifically failed map output.
**Why this works:** Distinguish full-loss and selective-loss epoch state.
For a same-epoch FetchFailed after selective loss, remove the named failed map
first and invoke executor or host cleanup with reliable-storage preservation
still enabled; reserve unselective bulk cleanup for failures without a prior
selective-loss state.
**Scope:** DAGScheduler cleanup epoch state and FetchFailed dispatch,
MapOutputTracker removal calls, and a mixed reliable/local scheduler regression
test.
**Compatibility:** The failed map output is still invalidated even when it
belongs to a reliable shuffle, local outputs remain removable, and unrelated
reliable shuffle registrations survive as promised by the new handle contract.
**Risks:** The selective-loss marker must be scoped by executor or host and
epoch so stale failures cannot affect later registrations. The named failed map
must be removed before preservation applies to the remaining reliable outputs.
**Constraints:** A FetchFailed for one shuffle must not be generalized into
evidence that every reliable shuffle on the executor was lost. Full cleanup
without a preceding selective loss must retain the existing broad invalidation
behavior.
**Success:** In an ExecutorLost followed by same-epoch FetchFailed sequence,
the failed map is unavailable and recomputed as needed while other reliable map
statuses from the executor remain registered.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4190,16 +4193,18 @@ private[spark] class DAGScheduler(
private[scheduler] def handleExecutorLost(
execId: String,
workerHost: Option[String]): Unit = {
- // if the cluster manager explicitly tells us that the entire worker was
lost, then
- // we know to unregister shuffle output. (Note that "worker" specifically
refers to the process
- // from a Standalone cluster, where the shuffle service lives in the
Worker.)
- val fileLost = !sc.shuffleDriverComponents.supportsReliableStorage() &&
- (workerHost.isDefined || !env.blockManager.externalShuffleServiceEnabled)
+ // Whether these outputs are candidates for removal at all; reliability is
then honored per
+ // shuffle via respectReliablyStored below. workerHost.isDefined means the
whole Standalone
+ // worker (which hosts the shuffle service) is gone.
+ val fileLost = workerHost.isDefined ||
!env.blockManager.externalShuffleServiceEnabled
removeExecutorAndUnregisterOutputs(
execId = execId,
fileLost = fileLost,
hostToUnregisterOutputs = workerHost,
- maybeEpoch = None)
+ maybeEpoch = None,
+ // Executor loss (not a fetch failure): preserve shuffles whose output
is reliably stored
Review Comment:
**Nit (P3):** The method Scaladoc still says all shuffle blocks for this
executor are assumed lost in the `fileLost` cases, but this selective call
intentionally retains every reliably stored status. Please qualify the contract
so only non-reliably-stored output is treated as lost during executor or worker
loss.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4334,7 +4344,9 @@ private[spark] class DAGScheduler(
message: String): Unit = {
logInfo(log"Shuffle files lost for worker ${MDC(WORKER_ID, workerId)} " +
log"on host ${MDC(HOST, host)}")
- mapOutputTracker.removeOutputsOnHost(host)
+ // Worker loss (not a fetch failure): reliably-stored shuffle output lives
off the worker and
Review Comment:
**Nit (P3):** The enclosing Scaladoc says all shuffle blocks on the removed
worker's host are lost and removed from `MapStatus`, while this call now
preserves reliable output. Please update it to say that only output not marked
reliably stored is dropped.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4267,7 +4272,8 @@ private[spark] class DAGScheduler(
fileLost: Boolean,
hostToUnregisterOutputs: Option[String],
maybeEpoch: Option[Long] = None,
- ignoreShuffleFileLostEpoch: Boolean = false): Unit = {
+ ignoreShuffleFileLostEpoch: Boolean = false,
Review Comment:
**Nit (P3):** This structured Scaladoc stops at `maybeEpoch`, so neither new
Boolean is described. Please add `@param` entries for
`ignoreShuffleFileLostEpoch` and `respectReliablyStored`, explaining separately
when epoch gating is bypassed and when reliable map statuses are preserved.
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1044,20 +1059,48 @@ 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).
+ *
+ * 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): Unit = {
- shuffleStatuses.valuesIterator.foreach { _.removeOutputsOnHost(host) }
- incrementEpoch()
+ def removeOutputsOnHost(host: String): Unit =
+ removeOutputsOnHost(host, respectReliablyStored = false)
+
+ def removeOutputsOnHost(host: String, respectReliablyStored: Boolean): Unit
= {
+ var removedAny = false
+ shuffleStatuses.valuesIterator.foreach { status =>
+ if (!(respectReliablyStored && status.isReliablyStored)) {
+ status.removeOutputsOnHost(host)
+ removedAny = true
Review Comment:
**Non-blocking (P2):** `removedAny` records that a shuffle was eligible for
cleanup, not that a map or merge status actually changed. In a mixed registry,
a reliable shuffle can own output on executor E while an unrelated local
shuffle has output only elsewhere; losing E removes nothing but still advances
the global epoch, so every executor clears and refetches identical map-status
data. The same mode/result confusion makes DAGScheduler omit its lost-epoch
stamp for ordinary shuffles even when no output was preserved, allowing a
same-epoch FetchFailed to repeat the full cleanup. Please return the actual
removed/preserved outcome from tracker cleanup, increment the epoch only for a
real metadata change, and drive the scheduler stamp from that outcome. Cover
both an ordinary-shuffle loss and the mixed-registry no-op boundary.
**Recommended change:** Make MapOutputTracker host and executor cleanup
return an aggregate outcome describing actual metadata removal and
reliable-output preservation, then use that outcome for both tracker epoch
increments and DAGScheduler lost-epoch bookkeeping.
**Why this works:** Have the per-shuffle removal operations report whether
map or merge state changed, aggregate those results while recording whether
reliable output was skipped, increment the global epoch only when state
changed, and stamp shuffleFileLostEpoch whenever cleanup was complete rather
than merely requested in selective mode.
**Scope:** ShuffleStatus and MapOutputTrackerMaster removal results,
DAGScheduler executor-loss bookkeeping, and focused MapOutputTrackerSuite and
DAGSchedulerSuite cases.
**Compatibility:** Legacy ordinary-shuffle loss remains full cleanup,
reliable shuffles remain preserved on executor or host loss, and the change
only avoids redundant invalidation when metadata did not change or nothing was
preserved.
**Risks:** Map and merge status removal must contribute consistently to the
aggregate result. Executor and host cleanup must retain identical epoch
semantics.
**Constraints:** Do not infer removal from shuffle reliability
classification. Do not suppress a later FetchFailed when reliable output was
actually preserved.
**Success:** The epoch advances exactly when tracker metadata changes,
ordinary full cleanup records its lost epoch, and a mixed-registry no-op
changes neither statuses nor epoch.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4267,7 +4272,8 @@ private[spark] class DAGScheduler(
fileLost: Boolean,
hostToUnregisterOutputs: Option[String],
maybeEpoch: Option[Long] = None,
- ignoreShuffleFileLostEpoch: Boolean = false): Unit = {
+ ignoreShuffleFileLostEpoch: Boolean = false,
+ respectReliablyStored: Boolean = false): Unit = {
Review Comment:
**Nit (P3):** Please add the missing `@param respectReliablyStored` contract
and qualify `fileLost`: when both are true, this helper does not assume all
associated shuffle blocks are gone; it preserves the reliable subset.
##########
core/src/main/scala/org/apache/spark/StreamingShuffleOutputTracker.scala:
##########
@@ -38,7 +38,12 @@ import org.apache.spark.util.ThreadUtils
*/
private[spark] trait ShuffleOutputTrackerMaster {
/** Register a shuffle so its outputs can be tracked. `jobId` is used by the
streaming tracker. */
- def registerShuffle(shuffleId: Int, numMaps: Int, numReduces: Int, jobId:
Int): Unit
+ def registerShuffle(
+ shuffleId: Int,
+ numMaps: Int,
+ numReduces: Int,
+ jobId: Int,
+ isReliablyStored: Boolean = false): Unit
Review Comment:
**Nit (P3):** `isReliablyStored` is now part of the common registration
surface, but the regular tracker persists it while this streaming
implementation accepts and ignores it. Please document that split on the common
member, including why pipelined output uses different loss handling, or keep
the regular-only argument out of the common contract.
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1044,20 +1059,48 @@ 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).
+ *
+ * When `respectReliablyStored` is true (executor/worker loss rather than a
fetch failure),
Review Comment:
**Nit (P3):** These paragraphs describe `respectReliablyStored`, but they
are attached to the one-argument wrappers, which always delegate with `false`.
Please attach the selective contract to the Boolean overloads and keep any
wrapper documentation explicit that those calls perform unconditional cleanup;
otherwise Scaladoc is misleading on the wrapper and absent on the callable
selective API.
--
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]
