Rename stageAttemptId to stageAttemptNumber

core/src/main/scala/org/apache/spark/TaskContext.scala

@@ -151,11 +151,11 @@ abstract class TaskContext extends Serializable {
   def stageId(): Int
 
   /**
-   * An ID that is unique to the stage attempt that this task belongs to. It represents how many
-   * times the stage has been attempted. The first stage attempt will be assigned stageAttemptId = 0
-   * , and subsequent attempts will increasing stageAttemptId one by one.
+   * How many times the stage that this task belongs to has been attempted. The first stage attempt
+   * will be assigned stageAttemptNumber = 0, and subsequent attempts will have increasing attempt
+   * numbers.
    */
-  def stageAttemptId(): Int
+  def stageAttemptNumber(): Int
 
   /**
    * The ID of the RDD partition that is computed by this task.

core/src/main/scala/org/apache/spark/TaskContextImpl.scala

@@ -42,7 +42,7 @@ import org.apache.spark.util._
  */
 private[spark] class TaskContextImpl(
     override val stageId: Int,
-    override val stageAttemptId: Int,
+    override val stageAttemptNumber: Int,
     override val partitionId: Int,
     override val taskAttemptId: Long,
     override val attemptNumber: Int,

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

@@ -79,7 +79,7 @@ private[spark] abstract class Task[T](
     SparkEnv.get.blockManager.registerTask(taskAttemptId)
     context = new TaskContextImpl(
       stageId,
-      stageAttemptId,
+      stageAttemptId, // stageAttemptId and stageAttemptNumber are semantically equal
       partitionId,
       taskAttemptId,
       attemptNumber,

core/src/test/java/test/org/apache/spark/JavaTaskContextCompileCheck.java

@@ -38,7 +38,7 @@ public static void test() {
     tc.attemptNumber();
     tc.partitionId();
     tc.stageId();
-    tc.stageAttemptId();
+    tc.stageAttemptNumber();
     tc.taskAttemptId();
   }
 
@@ -52,7 +52,7 @@ public void onTaskCompletion(TaskContext context) {
       context.isCompleted();
       context.isInterrupted();
       context.stageId();
-      context.stageAttemptId();
+      context.stageAttemptNumber();
       context.partitionId();
       context.addTaskCompletionListener(this);
     }

core/src/test/scala/org/apache/spark/memory/MemoryTestingUtils.scala

@@ -29,7 +29,7 @@ object MemoryTestingUtils {
     val taskMemoryManager = new TaskMemoryManager(env.memoryManager, 0)
     new TaskContextImpl(
       stageId = 0,
-      stageAttemptId = 0,
+      stageAttemptNumber = 0,
       partitionId = 0,
       taskAttemptId = 0,
       attemptNumber = 0,

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

@@ -159,28 +159,28 @@ class TaskContextSuite extends SparkFunSuite with BeforeAndAfter with LocalSpark
     assert(attemptIdsWithFailedTask.toSet === Set(0, 1))
   }
 
-  test("TaskContext.stageAttemptId getter") {
+  test("TaskContext.stageAttemptNumber getter") {
     sc = new SparkContext("local[1,2]", "test")
 
-    // Check stage attemptIds are 0 for initial stage
-    val stageAttemptIds = sc.parallelize(Seq(1, 2), 2).mapPartitions { _ =>
-      Seq(TaskContext.get().stageAttemptId()).iterator
+    // Check stageAttemptNumbers are 0 for initial stage
+    val stageAttemptNumbers = sc.parallelize(Seq(1, 2), 2).mapPartitions { _ =>
+      Seq(TaskContext.get().stageAttemptNumber()).iterator
     }.collect()
-    assert(stageAttemptIds.toSet === Set(0))
+    assert(stageAttemptNumbers.toSet === Set(0))
 
-    // Check stage attemptIds that are resubmitted when tasks have FetchFailedException
-    val stageAttemptIdsWithFailedStage =
+    // Check stageAttemptNumbers that are resubmitted when tasks have FetchFailedException
+    val stageAttemptNumbersWithFailedStage =
       sc.parallelize(Seq(1, 2, 3, 4), 4).repartition(1).mapPartitions { _ =>
-      val stageAttemptId = TaskContext.get().stageAttemptId()
-      if (stageAttemptId < 2) {
+      val stageAttemptNumber = TaskContext.get().stageAttemptNumber()
+      if (stageAttemptNumber < 2) {
         // Throw FetchFailedException to explicitly trigger stage resubmission. A normal exception
         // will only trigger task resubmission in the same stage.
         throw new FetchFailedException(null, 0, 0, 0, "Fake")
       }
-      Seq(stageAttemptId).iterator
+      Seq(stageAttemptNumber).iterator
     }.collect()
 
-    assert(stageAttemptIdsWithFailedStage.toSet === Set(2))
+    assert(stageAttemptNumbersWithFailedStage.toSet === Set(2))
   }
 
   test("accumulators are updated on exception failures") {

project/MimaExcludes.scala

@@ -37,7 +37,7 @@ object MimaExcludes {
   // Exclude rules for 2.3.x
   lazy val v23excludes = v22excludes ++ Seq(
     // [SPARK-22897] Expose stageAttemptId in TaskContext
-    ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.TaskContext.stageAttemptId"),
+    ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.TaskContext.stageAttemptNumber"),
 
     // SPARK-22789: Map-only continuous processing execution
     ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.spark.sql.streaming.StreamingQueryManager.startQuery$default$8"),

sql/core/src/test/scala/org/apache/spark/sql/execution/UnsafeFixedWidthAggregationMapSuite.scala

@@ -70,7 +70,7 @@ class UnsafeFixedWidthAggregationMapSuite
 
       TaskContext.setTaskContext(new TaskContextImpl(
         stageId = 0,
-        stageAttemptId = 0,
+        stageAttemptNumber = 0,
         partitionId = 0,
         taskAttemptId = Random.nextInt(10000),
         attemptNumber = 0,

sql/core/src/test/scala/org/apache/spark/sql/execution/UnsafeKVExternalSorterSuite.scala

@@ -116,7 +116,7 @@ class UnsafeKVExternalSorterSuite extends SparkFunSuite with SharedSQLContext {
     val taskMemMgr = new TaskMemoryManager(memoryManager, 0)
     TaskContext.setTaskContext(new TaskContextImpl(
       stageId = 0,
-      stageAttemptId = 0,
+      stageAttemptNumber = 0,
       partitionId = 0,
       taskAttemptId = 98456,
       attemptNumber = 0,