cloud-fan commented on code in PR #56559:
URL: https://github.com/apache/spark/pull/56559#discussion_r3507625448
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -958,6 +995,25 @@ private[spark] class MapOutputTrackerMaster(
incrementEpoch()
}
+ /**
+ * Mark a partition as having stale (redundant) push attempts and bump the
epoch so that
+ * reducers with a cached (empty) stale set are forced to re-fetch. Without
the epoch bump,
+ * a reducer that already fetched its merge statuses before this mark would
keep its stale
+ * set cached and never see the new mark, causing layer-3 fallback to
silently not fire.
+ *
+ * @param shuffleId the shuffle id.
+ * @param mapIndex the partition index (== mapIndex, not MapStatus.mapId) of
the stale
+ * (redundant) attempt; this is NOT MapStatus.mapId
+ */
+ def markStalePushedPartition(shuffleId: Int, mapIndex: Int): Unit = {
+ getShuffleStatusOrError(shuffleId, "markStalePushedPartition")
Review Comment:
`getShuffleStatusOrError` throws `ShuffleStatusNotFoundException` when the
shuffle is absent. This is now reached from
`TaskSetManager.detectStalePushIfShuffleTask` on the late/killed-attempt path
(`successful(index) && killedByOtherAttempt.contains(tid)`), and the exception
is `NonFatal`, so it propagates `handleSuccessfulTask` ->
`TaskSchedulerImpl.handleSuccessfulTask` -> `TaskResultGetter`'s `case
NonFatal(ex) => taskSetManager.abort(...)` -> `dagScheduler.taskSetFailed` —
i.e. a best-effort stale-detection optimization can fail the whole task set.
Before the round-4 refactor this was `shuffleStatuses.get(sid).foreach { ...
}`, a silent no-op when the shuffle was absent. The other
`getShuffleStatusOrError` writers (`registerMapOutput`, `unregisterMapOutput`,
...) only run on the DAGScheduler success/registration paths where the shuffle
is guaranteed registered; this new caller runs on an error-handling path where
the shuffle may already have been torn down (ContextCleaner GC after the
consuming stages finished, or a stage/job cancellation).
Non-blocking, but I'd make the mark tolerant of an absent shuffle, mirroring
the read accessors (`getShufflePushMergerLocations` -> `.getOrElse(...)`):
```suggestion
def markStalePushedPartition(shuffleId: Int, mapIndex: Int): Unit = {
shuffleStatuses.get(shuffleId).foreach { shuffleStatus =>
shuffleStatus.markStalePushedPartition(mapIndex)
// Bump the epoch so reducers with a cached stale set are forced to
re-fetch.
// A reducer that fetched its merge statuses before this mark
otherwise keeps
// its (empty) stale set and layer-3 fallback would not fire.
incrementEpoch()
}
}
```
Alternatively, keep the throwing form but guard at the call site as the old
code did. Is there an ordering guarantee that the shuffle is always still
registered when a late/killed attempt result arrives here? If so, a one-line
comment saying so would be enough.
--
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]
