(spark) branch branch-4.0 updated: [SPARK-59819][CORE] Fix `Resubmitted` task end accounting in `AppStatusListener`

[do...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dongjoon-hyun pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.0 by this push:
     new 167505c411d3 [SPARK-59819][CORE] Fix `Resubmitted` task end accounting in `AppStatusListener`
167505c411d3 is described below

commit 167505c411d38d9eacfeff2f9f2315417da6db37
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Mon Sep 28 07:34:42 2026 -0700

    [SPARK-59819][CORE] Fix `Resubmitted` task end accounting in `AppStatusListener`
    
    ### What changes were proposed in this pull request?
    
    This PR aims to fix two `Resubmitted` accounting bugs in `AppStatusListener.onTaskEnd`.
    
    When an executor is lost, `TaskSetManager.executorLost` posts `SparkListenerTaskEnd` with `Resubmitted` reason for the already-successful shuffle map tasks of that executor (without external shuffle service or on decommission) if the task set is not zombie. This event reuses the finished `TaskInfo` of the successful attempt and has no matching `SparkListenerTaskStart`. Since SPARK-41187 (apache/spark#38702), `onTaskEnd` uses `activeDelta = 0` for `Resubmitted`, but the following place [...]
    
    1. The speculation summary decrements `numActiveTasks` unconditionally. This PR uses `activeDelta` like the other active task counters.
    2. `ExecutorStageSummary.taskTime` and `ExecutorSummary.totalDuration` add `event.taskInfo.duration` again although the duration of the successful attempt was already counted. This PR skips them for `Resubmitted` like the executor-level task metrics, which are already skipped for `Resubmitted`.
    
    ### Why are the changes needed?
    
    To show correct values.
    
    - When a successful speculative task is resubmitted, `numActiveTasks` of the speculation summary becomes negative (e.g. `-1`), and nothing corrects it later because only `onTaskStart` and `onTaskEnd` update it. This value is stored in `SpeculationStageSummaryWrapper` and shown via the REST API `speculationSummary` field and the `Speculation Summary` table of the stage page.
    - For every `Resubmitted` task end, the duration of the successful attempt is counted twice in `ExecutorStageSummary.taskTime` (`Task Time` of `Aggregated Metrics by Executor` table in the stage page). `ExecutorSummary.totalDuration` (`Task Time (GC Time)` in the executors page and `totalDuration_seconds_total` in the Prometheus metrics) is also counted twice if the `Resubmitted` task end arrives before `SparkListenerExecutorRemoved` of the lost executor. This is timing-dependent beca [...]
    
    These were found during the review of apache/spark#58968.
    
    ### Does this PR introduce _any_ user-facing change?
    
    Yes, this fixes the incorrect values in the Spark UI, REST API, and Prometheus metrics. The negative `numActiveTasks` of the speculation summary exists since Apache Spark [v3.3.0](https://github.com/apache/spark/releases/tag/v3.3.0) (2022-06-09), and the double-counted task duration exists since Apache Spark [v2.3.0](https://github.com/apache/spark/releases/tag/v2.3.0) (2018-02-22).
    
    ### How was this patch tested?
    
    Pass the CIs with the newly added test case. I also manually verified that the new test case fails without this fix.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Claude Opus 5.5
    
    Closes #59092 from dongjoon-hyun/SPARK-59819.
    
    Authored-by: Dongjoon Hyun <do...@apache.org>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
    (cherry picked from commit b1b2685058d6359e00842cbd4c723f4ce3810709)
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../apache/spark/status/AppStatusListener.scala    |  9 +++--
 .../spark/status/AppStatusListenerSuite.scala      | 39 ++++++++++++++++++++++
 2 files changed, 45 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala b/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
index 5c93bf4bf77a..076a84bc2fea 100644
--- a/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
+++ b/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
@@ -741,7 +741,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
@@ -760,7 +763,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
@@ -782,7 +785,6 @@ 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
@@ -790,6 +792,7 @@ private[spark] class AppStatusListener(
       // 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
diff --git a/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala b/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala
index 4d75f5d7a1fc..cee2c362e9c9 100644
--- a/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala
@@ -1911,6 +1911,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)


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org