gaoyajun02 commented on code in PR #58008:
URL: https://github.com/apache/spark/pull/58008#discussion_r3851639166
##########
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:
##########
@@ -2921,11 +2921,49 @@ class TaskSetManagerSuite
s"\nCaptured logs:\n${logs.mkString("\n")}")
}
- test("SPARK-57491: late-arriving speculative ShuffleMapTask marks stale
partitionId") {
- sc = new SparkContext("local", "test")
+ test("SPARK-57491: late speculative ShuffleMapTask marks stale only when
fallback enabled") {
+ // (fallbackEnabled, detectAllStagesEnabled, expectStale)
+ Seq(
+ // Both switches enabled: the late attempt is marked stale regardless of
stage determinism
+ // (the test does not register an indeterminate ShuffleMapStage, so
detectAllStages is
+ // required here).
+ (true, true, true),
Review Comment:
Done. The matrix now takes `Option[Boolean]` so either switch can be left
unset:
- `(None, None, expectStale=false)` — exercises the real default (no
explicit config set); catches the default mismatch.
- `(Some(true), Some(false), expectStale=true,
runtimeIndeterminateAfterConstruction=true)` — enabled / detect-all-disabled,
with a registered `ShuffleMapStage` whose `isChecksumMismatched` is set *after*
`TaskSetManager` construction; catches the cached-snapshot issue (the
indeterminate-only path).
The helper registers a real `ShuffleMapStage` in
`dagScheduler.stageIdToStage` so the reference resolves, matching production.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -82,6 +82,19 @@ private[spark] class TaskSetManager(
private val isShuffleMapTasks = tasks(0).isInstanceOf[ShuffleMapTask]
// shuffleId is only available when isShuffleMapTasks=true
private val shuffleId = taskSet.shuffleId
+ // Scopes stale-push reducer fallback to indeterminate stages (deterministic
stages reproduce
+ // identical output across attempts, so a stale push there is benign).
Defaults to false when the
Review Comment:
Done. Reworded to "reproduce the same data set" — `UNORDERED` is not
indeterminate but permits record reordering, so "identical output" was stronger
than the invariant this guard relies on.
##########
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:
##########
@@ -2921,11 +2921,49 @@ class TaskSetManagerSuite
s"\nCaptured logs:\n${logs.mkString("\n")}")
}
- test("SPARK-57491: late-arriving speculative ShuffleMapTask marks stale
partitionId") {
- sc = new SparkContext("local", "test")
+ test("SPARK-57491: late speculative ShuffleMapTask marks stale only when
fallback enabled") {
+ // (fallbackEnabled, detectAllStagesEnabled, expectStale)
+ Seq(
+ // Both switches enabled: the late attempt is marked stale regardless of
stage determinism
+ // (the test does not register an indeterminate ShuffleMapStage, so
detectAllStages is
+ // required here).
+ (true, true, true),
+ // Fallback disabled (default): the late attempt is reported (logged)
but no map index is
+ // marked stale, so reducers keep reading the merged block.
+ (false, false, false)
+ ).foreach { case (fallbackEnabled, detectAllStagesEnabled, expectStale) =>
+ val staleMapIndexes =
+ runLateSpeculativeShuffleMapAttempt(fallbackEnabled,
detectAllStagesEnabled)
+ if (expectStale) {
+ assert(staleMapIndexes.contains(0),
+ s"Expected staleMapIndexes to contain mapIndex 0 " +
+ s"(fallback=$fallbackEnabled), got $staleMapIndexes")
+ } else {
+ assert(staleMapIndexes.isEmpty,
+ s"Expected no stale map indexes (fallback=$fallbackEnabled), got
$staleMapIndexes")
+ }
+ }
+ }
+
+ /**
+ * Drives a shuffle map stage through a late speculative attempt: task 0 and
task 1 start,
+ * task 1 finishes, task 0 is speculated, then the original task 0 finishes
(killing the
+ * speculative attempt) and finally the speculative attempt's result arrives
late. Returns
+ * the stale pushed map indexes recorded by the MapOutputTracker, which are
non-empty only
+ * when reducer fallback marking is enabled.
+ *
+ * Each invocation gets its own SparkContext via [[withSpark]] so the caller
can loop without
Review Comment:
Done. Changed `[[withSpark]]` -> `[[LocalSparkContext.withSpark]]`.
--
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]
