Refine

Author: Liupengcheng

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

@@ -181,9 +181,9 @@ private[spark] class OutputCommitCoordinator(conf: SparkConf, isDriver: Boolean)
       partition: Int,
       attemptNumber: Int): Boolean = synchronized {
     stageStates.get(stage) match {
-      case Some(state) if attemptInvalidOrFailed(state, stageAttempt, partition, attemptNumber) =>
+      case Some(state) if attemptOutdatedOrFailed(state, stageAttempt, partition, attemptNumber) =>
         logInfo(s"Commit denied for stage=$stage.$stageAttempt, partition=$partition: " +
-          s"task attempt $attemptNumber already marked as failed.")
+          s"task attempt $attemptNumber already outdated or marked as failed.")
         false
       case Some(state) =>
         val existing = state.authorizedCommitters(partition)
@@ -204,7 +204,7 @@ private[spark] class OutputCommitCoordinator(conf: SparkConf, isDriver: Boolean)
     }
   }
 
-  private def attemptInvalidOrFailed(
+  private def attemptOutdatedOrFailed(
       stageState: StageState,
       stageAttempt: Int,
       partition: Int,

core/src/test/scala/org/apache/spark/scheduler/OutputCommitCoordinatorSuite.scala

@@ -288,7 +288,7 @@ class OutputCommitCoordinatorSuite extends SparkFunSuite with BeforeAndAfter {
     outputCommitCoordinator.taskCompleted(stage, stageAttempt - 1, partition,
       attemptNumber = taskAttempt,
       reason = Success)
-    // attempts of current stage is authorized for committing
+    // attempts of latest retry stage is authorized for committing
     assert(outputCommitCoordinator.canCommit(stage, stageAttempt, partition, taskAttempt))
   }
 }