[SPARK-59819][CORE] Fix Resubmitted task end accounting in AppStatusListener

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala

@@ -766,7 +766,10 @@ private[spark] class AppStatusListener(
       }
 
       val esummary = stage.executorSummary(event.taskInfo.executorId)
-      esummary.taskTime += event.taskInfo.duration
+      // `Resubmitted` reuses the finished TaskInfo whose duration is already counted.
+      if (event.reason != Resubmitted) {
+        esummary.taskTime += event.taskInfo.duration
+      }
       esummary.succeededTasks += completedDelta
       esummary.failedTasks += failedDelta
       esummary.killedTasks += killedDelta
@@ -785,7 +788,7 @@ private[spark] class AppStatusListener(
       }
 
       if (event.taskInfo.speculative) {
-        stage.speculationStageSummary.numActiveTasks -= 1
+        stage.speculationStageSummary.numActiveTasks -= activeDelta
         stage.speculationStageSummary.numCompletedTasks += completedDelta
         stage.speculationStageSummary.numFailedTasks += failedDelta
         stage.speculationStageSummary.numKilledTasks += killedDelta
@@ -807,14 +810,14 @@ private[spark] class AppStatusListener(
       exec.activeTasks -= activeDelta
       exec.completedTasks += completedDelta
       exec.failedTasks += failedDelta
-      exec.totalDuration += event.taskInfo.duration
       exec.peakExecutorMetrics.compareAndUpdatePeakValues(event.taskExecutorMetrics)
 
       // Note: For resubmitted tasks, we continue to use the metrics that belong to the
       // first attempt of this task. This may not be 100% accurate because the first attempt
       // could have failed half-way through. The correct fix would be to keep track of the
       // metrics added by each attempt, but this is much more complicated.
       if (event.reason != Resubmitted) {
+        exec.totalDuration += event.taskInfo.duration
         if (event.taskMetrics != null) {
           val readMetrics = event.taskMetrics.shuffleReadMetrics
           exec.totalGcTime += event.taskMetrics.jvmGCTime

core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala

@@ -1979,6 +1979,45 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
     assert(listener.deadExecutors.size === 0)
   }
 
+  test("SPARK-59819: Resubmitted should not update speculative active tasks and task duration") {
+    val listener = new AppStatusListener(store, conf, true)
+    val appStore = new AppStatusStore(store)
+
+    listener.onExecutorAdded(createExecutorAddedEvent(1))
+    listener.onExecutorAdded(createExecutorAddedEvent(2))
+    val stage = new StageInfo(1, 0, "stage", 4, Nil, Nil, "details",
+      resourceProfileId = ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID)
+    listener.onJobStart(SparkListenerJobStart(1, time, Seq(stage), null))
+
+    time += 1
+    stage.submissionTime = Some(time)
+    listener.onStageSubmitted(SparkListenerStageSubmitted(stage, new Properties()))
+
+    val tasks = createTasks(2, Array("1", "2"))
+    tasks.foreach { task =>
+      listener.onTaskStart(SparkListenerTaskStart(stage.stageId, stage.attemptNumber(), task))
+    }
+    val speculativeTask = tasks(1)
+    assert(speculativeTask.speculative)
+
+    time += 1
+    speculativeTask.markFinished(TaskState.FINISHED, time)
+    listener.onTaskEnd(SparkListenerTaskEnd(stage.stageId, stage.attemptNumber(), "taskType",
+      Success, speculativeTask, new ExecutorMetrics, null))
+
+    // executor lost, success speculative task will be resubmitted with the same TaskInfo
+    time += 1
+    listener.onTaskEnd(SparkListenerTaskEnd(stage.stageId, stage.attemptNumber(), "taskType",
+      Resubmitted, speculativeTask, new ExecutorMetrics, null))
+
+    val execId = speculativeTask.executorId
+    assert(appStore.speculationSummary(stage.stageId, stage.attemptNumber())
+      .map(_.numActiveTasks) === Some(0))
+    assert(appStore.executorSummary(stage.stageId, stage.attemptNumber())(execId).taskTime ===
+      speculativeTask.duration)
+    assert(appStore.executorSummary(execId).totalDuration === speculativeTask.duration)
+  }
+
   test("SPARK-41683: Should correctly calculate numActiveStages if some stages are not submitted") {
     val stage1 = new StageInfo( 0, 0, "stage1", 0, Seq.empty, Seq.empty, "", resourceProfileId = 0)
     val stage2 = new StageInfo( 1, 0, "stage2", 0, Seq.empty, Seq.empty, "", resourceProfileId = 0)