[SPARK-45182][CORE] Ignore task completion from old stage after retrying indeterminate stages #42950

mayurdb · 2023-09-15T19:22:49Z

What changes were proposed in this pull request?

SPARK-25342 Added a support for rolling back shuffle map stage so that all tasks of the stage can be retried when the stage output is indeterminate. This is done by clearing all map outputs at the time of stage submission. This approach workouts well except for this case:

Assume both Shuffle 1 and 2 are indeterminate

ShuffleMapStage1 ––> Shuffle 1 ---–> ShuffleMapStage2 ----> Shuffle 2 ----> ResultStage

ShuffleMapStage1 is complete
A task from ShuffleMapStage2 fails with FetchFailed. Other tasks are still running
Both ShuffleMapStage1 and ShuffleMapStage2 are retried
ShuffleMapStage1 is retried and completes
ShuffleMapStage2 reattempt is scheduled for execution
Before all tasks of ShuffleMapStage2 reattempt could finish, one/more laggard tasks from the original attempt of ShuffleMapStage2 finish and ShuffleMapStage2 also gets marked as complete
Result Stage gets scheduled and finishes

After this change, such laggard tasks from the old attempt of the indeterminate stage will be ignored

Why are the changes needed?

This can give wrong result when indeterminate stages needs to be retried under the circumstances mentioned above

Does this PR introduce any user-facing change?

No

How was this patch tested?

A new test case

Was this patch authored or co-authored using generative AI tooling?

No

mayurdb · 2023-09-15T19:24:32Z

@cloud-fan @caican00 can you take a look?

mridulm

This is an important fix IMO, thanks for working on it @mayurdb

mridulm · 2023-09-19T18:37:14Z

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

-            shuffleStage.pendingPartitions -= task.partitionId
+            // Ignore task completion for old attempt of indeterminate stage
+            val ignoreIndeterminate = stage.isIndeterminate &&
+              task.stageAttemptId < stage.latestInfo.attemptNumber()


Why not wrap the entire case in if (!ignoreIndeterminate) { (and add a log message about it) ?

mridulm · 2023-09-19T18:37:58Z

+CC @Ngone51

cloud-fan · 2023-09-21T12:30:04Z

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -1903,13 +1903,20 @@ private[spark] class DAGScheduler(

          case smt: ShuffleMapTask =>
            val shuffleStage = stage.asInstanceOf[ShuffleMapStage]
-            shuffleStage.pendingPartitions -= task.partitionId


for deterministic stages, we do accept laggard tasks?

Yes, if determinate, this is fine.

cloud-fan · 2023-09-21T12:35:09Z

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

+              task.stageAttemptId < stage.latestInfo.attemptNumber()
+            if (!ignoreIndeterminate) {
+              shuffleStage.pendingPartitions -= task.partitionId
+            }
            val status = event.result.asInstanceOf[MapStatus]
            val execId = status.location.executorId
            logDebug("ShuffleMapTask finished on " + execId)
            if (executorFailureEpoch.contains(execId) &&
                smt.epoch <= executorFailureEpoch(execId)) {
              logInfo(s"Ignoring possibly bogus $smt completion from executor $execId")


not related to this PR, just noticed that we may mark the partition as completed but do not register the map output.

Agree, I am unsure about this.
+CC @Ngone51, @attilapiros who reviewed the original change - thoughts ?

I think this explains: https://github.com/apache/spark/pull/21976/files#diff-85de35b2e85646ed499c545a3be1cd3ffd525a88aae835a9c621f877eebadcb6L1385

We should probably add back that comment.

So this is kind of funny - take a look at what the above replaced @cloud-fan : https://github.com/apache/spark/pull/16620/files#diff-85de35b2e85646ed499c545a3be1cd3ffd525a88aae835a9c621f877eebadcb6R1183 :-)

Both of these actually do not account for DETERMINATE/INDETERMINATE changes we made subsequently.
IMO, for INDETERMINATE stages, we should ignore task completion events from previous attempts - since we have already cancelled the stage attempt.
Having said that, I have not thought through the nuances here.

+CC @jiangxb1987 as well

I also noticed this but seems to be fine since the missing partitions are defined by mapOutputTrackerMaster.findMissingPartitions rather than pendingPartitions.

If a task is finished but the map output is not registered, the missing partitions would be recomputed at the end of the stage, so this should be fine.

cloud-fan · 2023-09-25T03:35:28Z

LGTM if all tests pass

…ing indeterminate stages

jiangxb1987

LGTM

cloud-fan · 2023-09-26T03:03:38Z

thanks, merging to master/3.5!

…ing indeterminate stages ### What changes were proposed in this pull request? [SPARK-25342](https://issues.apache.org/jira/browse/SPARK-25342) Added a support for rolling back shuffle map stage so that all tasks of the stage can be retried when the stage output is indeterminate. This is done by clearing all map outputs at the time of stage submission. This approach workouts well except for this case: Assume both Shuffle 1 and 2 are indeterminate ShuffleMapStage1 ––> Shuffle 1 ---–> ShuffleMapStage2 ----> Shuffle 2 ----> ResultStage - ShuffleMapStage1 is complete - A task from ShuffleMapStage2 fails with FetchFailed. Other tasks are still running - Both ShuffleMapStage1 and ShuffleMapStage2 are retried - ShuffleMapStage1 is retried and completes - ShuffleMapStage2 reattempt is scheduled for execution - Before all tasks of ShuffleMapStage2 reattempt could finish, one/more laggard tasks from the original attempt of ShuffleMapStage2 finish and ShuffleMapStage2 also gets marked as complete - Result Stage gets scheduled and finishes After this change, such laggard tasks from the old attempt of the indeterminate stage will be ignored ### Why are the changes needed? This can give wrong result when indeterminate stages needs to be retried under the circumstances mentioned above ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? A new test case ### Was this patch authored or co-authored using generative AI tooling? No Closes #42950 from mayurdb/rollbackFix. Authored-by: mayurb <mayurb@uber.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit 7ffc0b7) Signed-off-by: Wenchen Fan <wenchen@databricks.com>

github-actions bot added the CORE label Sep 15, 2023

mridulm reviewed Sep 19, 2023

View reviewed changes

cloud-fan reviewed Sep 21, 2023

View reviewed changes

cloud-fan approved these changes Sep 25, 2023

View reviewed changes

mayurdb added 3 commits September 25, 2023 11:33

[SPARK-45182][CORE] Ignore task completion from old stage after retry…

c78882d

…ing indeterminate stages

Addressed comments

436443e

Reverse the condition to maintain git history

5880cf1

mayurdb force-pushed the rollbackFix branch from b37ee53 to 5880cf1 Compare September 25, 2023 06:05

Ngone51 approved these changes Sep 25, 2023

View reviewed changes

jiangxb1987 approved these changes Sep 25, 2023

View reviewed changes

cloud-fan closed this in 7ffc0b7 Sep 26, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-45182][CORE] Ignore task completion from old stage after retrying indeterminate stages #42950

[SPARK-45182][CORE] Ignore task completion from old stage after retrying indeterminate stages #42950

mayurdb commented Sep 15, 2023

mayurdb commented Sep 15, 2023

mridulm left a comment

mridulm Sep 19, 2023 •

edited

mridulm commented Sep 19, 2023

cloud-fan Sep 21, 2023

mridulm Sep 21, 2023

cloud-fan Sep 21, 2023

mridulm Sep 21, 2023

cloud-fan Sep 22, 2023

mridulm Sep 23, 2023 •

edited

Ngone51 Sep 25, 2023

jiangxb1987 Sep 25, 2023

cloud-fan commented Sep 25, 2023

jiangxb1987 left a comment

cloud-fan commented Sep 26, 2023

[SPARK-45182][CORE] Ignore task completion from old stage after retrying indeterminate stages #42950

[SPARK-45182][CORE] Ignore task completion from old stage after retrying indeterminate stages #42950

Conversation

mayurdb commented Sep 15, 2023

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

mayurdb commented Sep 15, 2023

mridulm left a comment

Choose a reason for hiding this comment

mridulm Sep 19, 2023 • edited

Choose a reason for hiding this comment

mridulm commented Sep 19, 2023

cloud-fan Sep 21, 2023

Choose a reason for hiding this comment

mridulm Sep 21, 2023

Choose a reason for hiding this comment

cloud-fan Sep 21, 2023

Choose a reason for hiding this comment

mridulm Sep 21, 2023

Choose a reason for hiding this comment

cloud-fan Sep 22, 2023

Choose a reason for hiding this comment

mridulm Sep 23, 2023 • edited

Choose a reason for hiding this comment

Ngone51 Sep 25, 2023

Choose a reason for hiding this comment

jiangxb1987 Sep 25, 2023

Choose a reason for hiding this comment

cloud-fan commented Sep 25, 2023

jiangxb1987 left a comment

Choose a reason for hiding this comment

cloud-fan commented Sep 26, 2023

mridulm Sep 19, 2023 •

edited

mridulm Sep 23, 2023 •

edited