Skip to content

Commit

Permalink
[SPARK-45498][CORE] Followup: Ignore task completion from old stage a…
Browse files Browse the repository at this point in the history
### What changes were proposed in this pull request?
With [SPARK-45182](https://issues.apache.org/jira/browse/SPARK-45182), we added a fix for not letting laggard tasks of the older attempts of the indeterminate stage from marking the partition has completed in the map output tracker.

When a task is completed, the DAG scheduler also notifies all the task sets of the stage about that partition being completed. Tasksets would not schedule such tasks if they are not already scheduled. This is not correct for the indeterminate stage, since we want to re-run all the tasks on a re-attempt

### Why are the changes needed?
Since the partition is not completed by older attempts and the partition from the newer attempt also doesn't get scheduled, the stage will have to be rescheduled to complete that partition. Since the stage is indeterminate, all the partitions will be recomputed

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Added check in existing unit test

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

Closes #43326 from mayurdb/indeterminateFix.

Authored-by: mayurb <mayurb@uber.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
  • Loading branch information
mayurdb authored and cloud-fan committed Oct 13, 2023
1 parent 12880c8 commit fb3b707
Show file tree
Hide file tree
Showing 2 changed files with 7 additions and 4 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -1847,9 +1847,9 @@ private[spark] class DAGScheduler(
case Success =>
// An earlier attempt of a stage (which is zombie) may still have running tasks. If these
// tasks complete, they still count and we can mark the corresponding partitions as
// finished. Here we notify the task scheduler to skip running tasks for the same partition,
// to save resource.
if (task.stageAttemptId < stage.latestInfo.attemptNumber()) {
// finished if the stage is determinate. Here we notify the task scheduler to skip running
// tasks for the same partition to save resource.
if (!stage.isIndeterminate && task.stageAttemptId < stage.latestInfo.attemptNumber()) {
taskScheduler.notifyPartitionCompletion(stageId, task.partitionId)
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3169,13 +3169,16 @@ class DAGSchedulerSuite extends SparkFunSuite with TempLocalSparkContext with Ti
makeMapStatus("hostB",
2)))

// The second task of the shuffle map stage 1 from 1st attempt succeeds
// The second task of the shuffle map stage 1 from 1st attempt succeeds
runEvent(makeCompletionEvent(
taskSets(1).tasks(1),
Success,
makeMapStatus("hostC",
2)))

// Above task completion should not mark the partition 1 complete from 2nd attempt
assert(!tasksMarkedAsCompleted.contains(taskSets(3).tasks(1)))

// This task completion should get ignored and partition 1 should be missing
// for shuffle map stage 1
assert(mapOutputTracker.findMissingPartitions(shuffleId2) == Some(Seq(1)))
Expand Down

0 comments on commit fb3b707

Please sign in to comment.