This is an automated email from the ASF dual-hosted git repository.
sunchao pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new eb2177566299 [SPARK-59094][CORE] Ignore late successes from
invalidated barrier shuffle attempts
eb2177566299 is described below
commit eb21775662994111c93f6a8d7adb1b4912249cac
Author: Chao Sun <[email protected]>
AuthorDate: Mon Aug 31 09:29:54 2026 -0700
[SPARK-59094][CORE] Ignore late successes from invalidated barrier shuffle
attempts
### What changes were proposed in this pull request?
Fix [SPARK-59094](https://issues.apache.org/jira/browse/SPARK-59094) by
ignoring successful completions from failed barrier shuffle attempts before
they can register output or mark a replacement task set's partition complete.
Track barrier failures in `latestFailedBarrierAttemptId`, separate from
indeterminate rollback bookkeeping. This preserves rollback cleanup when
relocation restores an output before retry. Valid replacement successes and
ordinary deterministic, non-barrier output reuse are unchanged.
Producer-side output clearing and asynchronous migration updates are
unchanged. The separate case of a late consumer failure clearing an active
producer retry's accepted output remains for a follow-up JIRA, which has not
yet been filed.
### Why are the changes needed?
Barrier recovery must rerun every participant. Accepting a late success
from a failed attempt can omit a partition from the retry or prevent the
replacement task set from launching.
### Does this PR introduce _any_ user-facing change?
Yes. Barrier retries retain every required participant in the affected
event orderings. No public APIs or configuration options change.
### How was this patch tested?
The eleven added scheduler cases cover both barrier failure paths,
before/after resubmission, repeated failed attempts, valid retry results,
ordinary non-barrier output reuse, and indeterminate rollback cleanup after
relocation. The producer-retry control checks acceptance of subsequent
successes after a late consumer fetch failure.
- `DAGSchedulerSuite`: 233 tests passed.
- `TaskSchedulerImplSuite`: 121 tests passed.
Both suites were run with ScalaTest after Maven test compilation.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex
This contribution is original work, licensed to the project under the
Apache License, Version 2.0.
Closes #58402 from
sunchao/dev/chao/codex/barrier-retry-ignore-old-results-20260828.
Authored-by: Chao Sun <[email protected]>
Signed-off-by: Chao Sun <[email protected]>
(cherry picked from commit 7fa054b411af6d2d59b4e860e3d316af9e8594c0)
Signed-off-by: Chao Sun <[email protected]>
---
.../org/apache/spark/scheduler/DAGScheduler.scala | 13 +-
.../scala/org/apache/spark/scheduler/Stage.scala | 6 +
.../apache/spark/scheduler/DAGSchedulerSuite.scala | 200 +++++++++++++++++++++
3 files changed, 215 insertions(+), 4 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 67b7823091e2..6a897e461528 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -3338,10 +3338,10 @@ private[spark] class DAGScheduler(
// finished if the stage is determinate. Here we notify the task
scheduler to skip running
// tasks for the same partition to save resource.
- // Ignore task completion for old attempt of stages with
nondeterministic output.
- // This is tracked via maxAttemptIdToIgnore which is set when a stage
is rolled back.
+ // Ignore task completion from attempts invalidated by rollback or
barrier stage failure.
val ignoreOldTaskAttempts =
- stage.maxAttemptIdToIgnore.exists(_ >= task.stageAttemptId)
+ stage.maxAttemptIdToIgnore.exists(_ >= task.stageAttemptId) ||
+ stage.latestFailedBarrierAttemptId.exists(_ >= task.stageAttemptId)
if (!ignoreOldTaskAttempts && task.stageAttemptId <
stage.latestInfo.attemptNumber()) {
taskScheduler.notifyPartitionCompletion(stageId, task.partitionId)
@@ -3475,7 +3475,8 @@ private[spark] class DAGScheduler(
}
}
} else {
- logInfo(log"Ignoring ${MDC(TASK_NAME, smt)} completion from an
older attempt of indeterminate stage")
+ logInfo(log"Ignoring ${MDC(TASK_NAME, smt)} completion from " +
+ log"an invalidated stage attempt")
}
if (runningStages.contains(shuffleStage) &&
shuffleStage.pendingPartitions.isEmpty) {
@@ -3601,6 +3602,8 @@ private[spark] class DAGScheduler(
if (failedStage.rdd.isBarrier()) {
failedStage match {
case failedMapStage: ShuffleMapStage =>
+ // Ignore late completions so the replacement barrier stage
reruns every task.
+ failedMapStage.latestFailedBarrierAttemptId =
Some(task.stageAttemptId)
// Mark all the map as broken in the map stage, to ensure
retry all the tasks on
// resubmitted stage attempt.
mapOutputTracker.unregisterAllMapAndMergeOutput(
@@ -3709,6 +3712,8 @@ private[spark] class DAGScheduler(
} else {
failedStage match {
case failedMapStage: ShuffleMapStage =>
+ // Ignore late completions so the replacement barrier stage
reruns every task.
+ failedMapStage.latestFailedBarrierAttemptId =
Some(task.stageAttemptId)
// Mark all the map as broken in the map stage, to ensure
retry all the tasks on
// resubmitted stage attempt.
mapOutputTracker.unregisterAllMapAndMergeOutput(failedMapStage.shuffleDep.shuffleId)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/Stage.scala
b/core/src/main/scala/org/apache/spark/scheduler/Stage.scala
index f22776bbc196..96ee1a8ec598 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/Stage.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/Stage.scala
@@ -92,6 +92,12 @@ private[scheduler] abstract class Stage(
*/
private[scheduler] var maxAttemptIdToIgnore: Option[Int] = None
+ /**
+ * The latest failed barrier attempt. Results from this and older attempts
must be ignored.
+ * Keep this separate from maxAttemptIdToIgnore, which also deduplicates
rollback cleanup.
+ */
+ private[scheduler] var latestFailedBarrierAttemptId: Option[Int] = None
+
val name: String = callSite.shortForm
val details: String = callSite.longForm
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
index 518347b1b090..3a824efec511 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
@@ -1565,6 +1565,206 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
assertDataStructuresEmpty()
}
+ for (completeBeforeRetry <- Seq(true, false); failedAttempts <- 1 to 2) {
+ val completionTime = if (completeBeforeRetry) "before" else "after"
+ test("SPARK-59094: Ignore successful tasks from failed barrier attempts " +
+ s"$completionTime resubmission (failedAttempts=$failedAttempts)") {
+ val mapRdd = new MyRDD(sc, 2, Nil).barrier().mapPartitions(iter => iter)
+ val shuffleDep = new ShuffleDependency(mapRdd, new HashPartitioner(2))
+ val resultRdd = new MyRDD(sc, 2, List(shuffleDep), tracker =
mapOutputTracker)
+ submit(resultRdd, Array(0, 1))
+ val mapStage = scheduler.shuffleIdToMapStage(shuffleDep.shuffleId)
+ assert(!mapStage.isStaticallyIndeterminate)
+
+ val originalAttempts = (0 until failedAttempts).map { attempt =>
+ if (attempt > 0) {
+ scheduler.resubmitFailedStages()
+ }
+ val taskSetToFail = taskSet(0, attempt)
+ // Process completions directly so queued retry events cannot change
their ordering.
+
scheduler.handleTaskCompletion(makeCompletionEvent(taskSetToFail.tasks(0),
+ new ExceptionFailure(new RuntimeException("barrier task failed"),
Seq.empty), null))
+ taskSetToFail
+ }
+ assert(mapOutputTracker.findMissingPartitions(shuffleDep.shuffleId) ===
Some(Seq(0, 1)))
+
+ if (!completeBeforeRetry) {
+ scheduler.resubmitFailedStages()
+ }
+ assert(mapStage.latestInfo.attemptNumber() ===
+ (if (completeBeforeRetry) failedAttempts - 1 else failedAttempts))
+ // The other task finished before cancellation, but its success reached
the driver late.
+ originalAttempts.foreach { originalAttempt =>
+ scheduler.handleTaskCompletion(makeCompletionEvent(
+ originalAttempt.tasks(1), Success, makeMapStatus("hostB", 2)))
+ }
+ assert(mapOutputTracker.findMissingPartitions(shuffleDep.shuffleId) ===
Some(Seq(0, 1)))
+ assert(tasksMarkedAsCompleted.isEmpty)
+ if (!completeBeforeRetry) {
+ assert(mapStage.pendingPartitions.toSet === Set(0, 1))
+ }
+
+ if (completeBeforeRetry) {
+ scheduler.resubmitFailedStages()
+ }
+ assert(taskSet(0, failedAttempts).tasks.map(_.partitionId).toSeq ===
Seq(0, 1))
+ completeShuffleMapStageSuccessfully(0, failedAttempts, 2)
+ assert(taskSets.count(_.stageId == 0) === failedAttempts + 1)
+ completeNextResultStageWithSuccess(1, 0)
+ assert(results === Map(0 -> 42, 1 -> 42))
+ assertDataStructuresEmpty()
+ }
+ }
+
+ for (barrier <- Seq(true, false); completeBeforeRetry <- Seq(true, false)) {
+ val completionTime = if (completeBeforeRetry) "before" else "after"
+ test(s"SPARK-59094: Late success $completionTime resubmission after a
shuffle fetch failure " +
+ s"in a barrier=$barrier map stage") {
+ // The middle stage reads one shuffle and produces a different shuffle.
+ val inputRdd = new MyRDD(sc, 2, Nil)
+ val inputDep = new ShuffleDependency(inputRdd, new HashPartitioner(2))
+ val middleRdd = new MyRDD(sc, 2, List(inputDep), tracker =
mapOutputTracker)
+ val mapRdd = if (barrier) middleRdd.barrier().mapPartitions(iter =>
iter) else middleRdd
+ val outputDep = new ShuffleDependency(mapRdd, new HashPartitioner(2))
+ val resultRdd = new MyRDD(sc, 2, List(outputDep), tracker =
mapOutputTracker)
+ submit(resultRdd, Array(0, 1))
+ completeShuffleMapStageSuccessfully(0, 0, 2)
+ val originalAttempt = taskSet(1, 0)
+
assert(!scheduler.shuffleIdToMapStage(outputDep.shuffleId).isStaticallyIndeterminate)
+
+ // Process completions directly so queued retry events cannot change
their ordering.
+
scheduler.handleTaskCompletion(makeCompletionEvent(originalAttempt.tasks(0),
+ FetchFailed(makeBlockManagerId("hostA"), inputDep.shuffleId, 0L, 0, 0,
"ignored"), null))
+
+ def retryParentStage(): Unit = {
+ scheduler.resubmitFailedStages()
+ completeShuffleMapStageSuccessfully(0, 1, 2, Seq("hostD"))
+ }
+ if (!completeBeforeRetry) {
+ retryParentStage()
+ }
+
assert(scheduler.shuffleIdToMapStage(outputDep.shuffleId).latestInfo.attemptNumber()
===
+ (if (completeBeforeRetry) 0 else 1))
+ // This output is on a healthy executor, so executor-loss filtering
cannot hide the race.
+ scheduler.handleTaskCompletion(makeCompletionEvent(
+ originalAttempt.tasks(1), Success, makeMapStatus("hostC", 2)))
+ val missingPartitions = if (barrier) Seq(0, 1) else Seq(0)
+ assert(mapOutputTracker.findMissingPartitions(outputDep.shuffleId) ===
+ Some(missingPartitions))
+ if (barrier || completeBeforeRetry) {
+ assert(tasksMarkedAsCompleted.isEmpty)
+ } else {
+ assert(tasksMarkedAsCompleted.toSeq ===
+ Seq(taskSet(1, 1).tasks.find(_.partitionId == 1).get))
+ }
+ if (barrier && !completeBeforeRetry) {
+
assert(scheduler.shuffleIdToMapStage(outputDep.shuffleId).pendingPartitions.toSet
===
+ Set(0, 1))
+ }
+
+ if (completeBeforeRetry) {
+ retryParentStage()
+ }
+ val retry = taskSet(1, 1)
+ val submittedPartitions = if (completeBeforeRetry) missingPartitions
else Seq(0, 1)
+ assert(retry.tasks.map(_.partitionId).toSeq === submittedPartitions)
+ // An ordinary determinate stage may reuse the old success. A barrier
stage must rerun both.
+ retry.tasks.filter(task =>
missingPartitions.contains(task.partitionId)).foreach { task =>
+ runEvent(makeCompletionEvent(task, Success, makeMapStatus("hostD", 2)))
+ }
+ assert(taskSets.count(_.stageId == 1) === 2)
+ completeNextResultStageWithSuccess(2, 0)
+ assert(results === Map(0 -> 42, 1 -> 42))
+ assertDataStructuresEmpty()
+ }
+ }
+
+ test("SPARK-59094: accept barrier retry successes after a late consumer
fetch failure") {
+ val mapRdd = new MyRDD(sc, 2, Nil).barrier().mapPartitions(iter => iter)
+ val shuffleDep = new ShuffleDependency(mapRdd, new HashPartitioner(2))
+ val resultRdd = new MyRDD(sc, 2, List(shuffleDep), tracker =
mapOutputTracker)
+ submit(resultRdd, Array(0, 1))
+ completeShuffleMapStageSuccessfully(0, 0, 2)
+ val originalResultAttempt = taskSet(1, 0)
+
+ runEvent(makeCompletionEvent(originalResultAttempt.tasks(0),
+ FetchFailed(makeBlockManagerId("hostA"), shuffleDep.shuffleId, 0L, 0, 0,
"ignored"), null))
+ scheduler.resubmitFailedStages()
+ assert(taskSet(0, 1).tasks.map(_.partitionId).toSeq === Seq(0, 1))
+
+ // Another task from the same consumer attempt can fail while the producer
is being retried.
+ runEvent(makeCompletionEvent(originalResultAttempt.tasks(1),
+ FetchFailed(makeBlockManagerId("hostB"), shuffleDep.shuffleId, 1L, 1, 1,
"ignored"), null))
+ scheduler.resubmitFailedStages()
+ assert(taskSets.count(_.stageId == 0) === 2)
+
+ completeShuffleMapStageSuccessfully(0, 1, 2, Seq("hostC", "hostD"))
+ completeNextResultStageWithSuccess(1, 1)
+ assert(results === Map(0 -> 42, 1 -> 42))
+ assertDataStructuresEmpty()
+ }
+
+ for (fetchFailure <- Seq(false, true)) {
+ val failureType = if (fetchFailure) "fetch failure" else "task failure"
+ test(s"SPARK-59094: retry all indeterminate barrier tasks after
$failureType and relocation") {
+ val inputRdd = new MyRDD(sc, 2, Nil)
+ val inputDep = new ShuffleDependency(inputRdd, new HashPartitioner(2))
+ val mapRdd = new MyRDD(sc, 2, List(inputDep), tracker = mapOutputTracker,
+ indeterminate = true).barrier().mapPartitions(iter => iter)
+ val outputDep = new ShuffleDependency(mapRdd, new HashPartitioner(2))
+ val resultRdd = new MyRDD(sc, 2, List(outputDep), tracker =
mapOutputTracker)
+ submit(resultRdd, Array(0, 1))
+
assert(!scheduler.shuffleIdToMapStage(inputDep.shuffleId).isStaticallyIndeterminate)
+ complete(taskSet(0, 0), Seq(
+ (Success, makeMapStatus("hostA", 2, mapTaskId = 100L)),
+ (Success, makeMapStatus("hostB", 2, mapTaskId = 101L))))
+ val mapStage = scheduler.shuffleIdToMapStage(outputDep.shuffleId)
+ assert(mapStage.isStaticallyIndeterminate)
+ assert(!outputDep.checksumMismatchFullRetryEnabled)
+ val originalAttempt = taskSet(1, 0)
+ val completedStatus = makeMapStatus("hostC", 2, mapTaskId = 200L)
+
+ // Keep the failure and relocation ahead of any queued retry event.
+ scheduler.handleTaskCompletion(makeCompletionEvent(
+ originalAttempt.tasks(0), Success, completedStatus))
+ assert(mapOutputTracker.findMissingPartitions(outputDep.shuffleId) ===
Some(Seq(1)))
+ val failureReason: TaskFailedReason = if (fetchFailure) {
+ FetchFailed(makeBlockManagerId("hostA"), inputDep.shuffleId, 100L, 0,
1, "ignored")
+ } else {
+ new ExceptionFailure(new RuntimeException("barrier task failed"),
Seq.empty)
+ }
+ scheduler.handleTaskCompletion(makeCompletionEvent(
+ originalAttempt.tasks(1), failureReason, null))
+ assert(mapOutputTracker.findMissingPartitions(outputDep.shuffleId) ===
Some(Seq(0, 1)))
+
+ // A delayed migration report can restore an output cleared by the
barrier failure.
+ val relocatedHost = makeBlockManagerId("hostD")
+ mapOutputTracker.updateMapOutput(outputDep.shuffleId, 200L,
relocatedHost)
+ assert(completedStatus.location === relocatedHost)
+ assert(mapOutputTracker.findMissingPartitions(outputDep.shuffleId) ===
Some(Seq(1)))
+ assert(mapStage.latestInfo.attemptNumber() === 0)
+
+ scheduler.resubmitFailedStages()
+ if (fetchFailure) {
+ val parentRetry = taskSet(0, 1)
+ assert(parentRetry.tasks.map(_.partitionId).toSeq === Seq(0))
+ complete(parentRetry, Seq((Success, makeMapStatus("hostE", 2,
mapTaskId = 102L))))
+ }
+ val retry = taskSet(1, 1)
+ assert(retry.tasks.map(_.partitionId).toSeq === Seq(0, 1))
+ assert(retry.tasks.forall(task => task.isBarrier && task.numPartitions
== 2))
+ assert(mapStage.pendingPartitions.toSet === Set(0, 1))
+ assert(mapOutputTracker.findMissingPartitions(outputDep.shuffleId) ===
Some(Seq(0, 1)))
+ complete(retry, Seq(
+ (Success, makeMapStatus("hostF", 2, mapTaskId = 201L)),
+ (Success, makeMapStatus("hostG", 2, mapTaskId = 202L))))
+ assert(taskSets.count(_.stageId == 1) === 2)
+ completeNextResultStageWithSuccess(2, 0)
+ assert(results === Map(0 -> 42, 1 -> 42))
+ assertDataStructuresEmpty()
+ }
+ }
+
test("SPARK-58887: a barrier job can be cancelled while its slot check is
being retried") {
// 3 barrier tasks on the local[2] backend fail the max concurrent tasks
check, so the
// submission enters the retry window during which the job is registered
nowhere but
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
