tgravescs commented on a change in pull request #34735:
URL: https://github.com/apache/spark/pull/34735#discussion_r765951281
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -718,6 +718,12 @@ private[spark] class DAGScheduler(
// read from merged output as the MergeStatuses are not
available.
if (!mapStage.isAvailable ||
!mapStage.shuffleDep.shuffleMergeFinalized) {
missing += mapStage
+ } else if (mapStage.nextAttemptId == 0) {
+ // Forward the nextAttemptId if skipped and get visited for
the first time.
+ // Otherwise, once it gets retried,
+ // 1) the stuffs in stage info become distorting, e.g. task
num, input byte, e.t.c
+ // 2) the first attempt starts from 0-idx, it will not be
marked as a retry
+ mapStage.nextAttemptId += 1
Review comment:
its incrementing nextAttemptId (ie the next attempt, not the current
one). The skipped attempt never calls makeNewStageAttempt so the first time it
was skipped this was never incremented. The default of nextAttemptId on Stage
creation is 0 and since it was never incremented in the skipped stage when you
call makeNewStageAttempt for the failed attempt it is run as 0 but the next
attempt would be 1. So the skipped attempt and the rerun failed attempt both
are attempt 0.
--
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]
