otterc commented on a change in pull request #30691: URL: https://github.com/apache/spark/pull/30691#discussion_r640073914
########## File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala ########## @@ -2136,9 +2137,24 @@ private[spark] class DAGScheduler( } } - private[scheduler] def handleShuffleMergeFinalized(stage: ShuffleMapStage): Unit = { - stage.shuffleDep.markShuffleMergeFinalized - processShuffleMapStageCompletion(stage) + private[scheduler] def handleRegisterMergeStatuses( + stage: ShuffleMapStage, + mergeStatuses: Seq[(Int, MergeStatus)]): Unit = { + // Register merge statuses if the stage is still running and shuffle merge is not finalized yet. + if (runningStages.contains(stage) && !stage.shuffleDep.shuffleMergeFinalized) { + mapOutputTracker.registerMergeResults(stage.shuffleDep.shuffleId, mergeStatuses) + } + } + + private[scheduler] def handleShuffleMergeFinalized( + stage: ShuffleMapStage): Unit = { + // Only update MapOutputTracker metadata if the stage is still active. i.e not cancelled. + if (runningStages.contains(stage)) { + stage.shuffleDep.markShuffleMergeFinalized() + processShuffleMapStageCompletion(stage) + } else { + mapOutputTracker.unregisterAllMergeResult(stage.shuffleDep.shuffleId) Review comment: > Stage will be active until the shuffle merge is finalized and then only we are processing the map stage completion, isn't it? So if the stage is not part of running stages and we still reach the handling shuffle merge finalize, then we need to unregister the merge results, isn't it? Are you just making an assumption here, that if the stage is not running and then this finalized message is processed that means the stage is cancelled? Is this a valid assumption? If yes, then can you add a comment here. > Can you think of a scenario where stage is not part of running stages and still shuffle merge is finalized? - Ideally this should not happen. This is what is throwing me off. If this is not ideally going to happen then why are we unregistering the results here. Again, if the assumption is that this happens when the stage was cancelled then document it. Also, if handling stage cancellation wrt merge finalization is not handled in this PR then why have this unregistration of merge results here? -- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org