@Ngone51's comments

Author: dagrawal3409

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

@@ -34,6 +34,6 @@ case class ExecutorDecommissionInfo(message: String, isHostDecommissioned: Boole
  */
 case class ExecutorDecommissionState(
     message: String,
-    // Timestamp in milliseconds when decommissioning was triggered
-    tsMillis: Long,
+    // Timestamp the decommissioning commenced in millis since epoch of the driver's clock
+    startTime: Long,
     isHostDecommissioned: Boolean)

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

@@ -930,7 +930,7 @@ private[spark] class TaskSchedulerImpl(
         if (!oldDecomState.exists(_.isHostDecommissioned)) {
           executorsPendingDecommission(executorId) = ExecutorDecommissionState(
             decommissionInfo.message,
-            oldDecomState.map(_.tsMillis).getOrElse(clock.getTimeMillis()),
+            oldDecomState.map(_.startTime).getOrElse(clock.getTimeMillis()),
             decommissionInfo.isHostDecommissioned)
         }
       }

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

@@ -1056,11 +1056,13 @@ private[spark] class TaskSetManager(
           val taskInfo = taskInfos(tid)
           val decomState = sched.getExecutorDecommissionState(taskInfo.executorId)
           if (decomState.nonEmpty) {
-            // Check whether this task will finish before the exectorKillTime assuming
-            // it will take medianDuration overall. If this task cannot finish within
-            // executorKillInterval, then this task is a candidate for speculation
+            // Check if this task might finish after this executor is decommissioned.
+            // We estimate the task's finish time by using the median task duration.
+            // Whereas the time when the executor might be decommissioned is estimated using the
+            // config executorDecommissionKillInterval. If the task is going to finish after
+            // decommissioning, then we will eagerly speculate the task.
             val taskEndTimeBasedOnMedianDuration = taskInfos(tid).launchTime + medianDuration
-            val executorDecomTime = decomState.get.tsMillis + executorDecommissionKillInterval.get
+            val executorDecomTime = decomState.get.startTime + executorDecommissionKillInterval.get
             val canExceedDeadline = executorDecomTime < taskEndTimeBasedOnMedianDuration
             if (canExceedDeadline) {
               speculated = checkAndSubmitSpeculatableTask(tid, time, 0)

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

@@ -1977,7 +1977,7 @@ class TaskSetManagerSuite
     // (TASK 2 -> 15, TASK 3 -> 15)
     sched.executorDecommission("exec2", ExecutorDecommissionInfo("decom",
       isHostDecommissioned = false))
-    assert(sched.getExecutorDecommissionState("exec2").map(_.tsMillis) ===
+    assert(sched.getExecutorDecommissionState("exec2").map(_.startTime) ===
       Some(clock.getTimeMillis()))
 
     assert(manager.checkSpeculatableTasks(0))