update

Author: shahidki31

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

@@ -190,8 +190,9 @@ private class LiveTask(
     val handleZeros = mutable.HashSet[String]()
 
     /**
-     * For non successful tasks, store the metrics as negetive to avoid the calculation in the
-     * task summary. `toApi` method in TaskDataWrapper will make it actual value.
+     * SPARK-26260: For non successful tasks, store the metrics as negetive to avoid
+     * the calculation in the task summary. `toApi` method in the `TaskDataWrapper` will make
+     * it actual value.
      */
     val taskMetrics: v1.TaskMetrics = if (hasMetrics && !info.successful) {
       makeNegative(metrics, handleZeros)

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

@@ -181,7 +181,7 @@ private[spark] class TaskDataWrapper(
     val errorMessage: Option[String],
 
     val hasMetrics: Boolean,
-    // Non successful metrics will have negative values in `TaskDataWrapper`.
+    // Non successful metrics now will have negative values in `TaskDataWrapper`.
     // `TaskData` will have actual metric values. To recover the actual metric value
     // from `TaskDataWrapper`, need use `getMetricValue` method. parameter `handleZero` is to
     // check whether the index has zero metric value, which is used in the `getMetricValue`.
@@ -242,9 +242,9 @@ private[spark] class TaskDataWrapper(
     val stageId: Int,
     val stageAttemptId: Int) {
 
-  // To handle non successful tasks metrics (Running, Failed, Killed).
+  // SPARK-26260: To handle non successful tasks metrics (Running, Failed, Killed).
   private def getMetricValue(metric: Long, index: String): Long = {
-    if (handleZero(index)) {
+    if (status != "SUCCESS" && handleZero(index)) {
       0L
     } else {
       math.abs(metric)

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

@@ -133,15 +133,6 @@ class AppStatusStoreSuite extends SparkFunSuite {
     }
   }
 
-  test("task summary size for default metrics should be zero") {
-    val store = new InMemoryStore()
-    (0 until 5).foreach { _ => store.write(newTaskData(-1, status = "RUNNING")) }
-    Seq(new AppStatusStore(store), createLiveStore(store)).foreach { appStore =>
-      val summary = appStore.taskSummary(stageId, attemptId, uiQuantiles)
-      assert(summary.size === 0)
-    }
-  }
-
   test("SPARK-26260: summary should contain successful tasks only when with LevelDB store") {
     val testDir = Utils.createTempDir()
     val diskStore = KVUtils.open(testDir, getClass().getName())
@@ -168,6 +159,15 @@ class AppStatusStoreSuite extends SparkFunSuite {
     Utils.deleteRecursively(testDir)
   }
 
+  test("SPARK-26260: task summary size for default metrics should be zero") {
+    val store = new InMemoryStore()
+    store.write(newTaskData(-1, status = "RUNNING"))
+    Seq(new AppStatusStore(store), createLiveStore(store)).foreach { appStore =>
+      val summary = appStore.taskSummary(stageId, attemptId, uiQuantiles)
+      assert(summary.size === 0)
+    }
+  }
+
   private def compareQuantiles(count: Int, quantiles: Array[Double]): Unit = {
     val store = new InMemoryStore()
     val values = (0 until count).map { i =>