[SPARK-10515] When killing executor, the pending replacement executors should not be lost

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -66,6 +66,12 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   // Executors we have requested the cluster manager to kill that have not died yet
   private val executorsPendingToRemove = new HashSet[String]
 
+  // Executors we have requested the cluster manager to replace with new ones that have killed
+  private val executorsToReplace = new HashSet[String]
+
+  // Number of executors requested from the cluster manager that have not replaced yet
+  private var numReplacingExecutors = 0
+
   // A map to store hostname with its possible task number running on it
   protected var hostToLocalTaskCount: Map[String, Int] = Map.empty
 
@@ -147,6 +153,10 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
               numPendingExecutors -= 1
               logDebug(s"Decremented number of pending executors ($numPendingExecutors left)")
             }
+            if (numReplacingExecutors > 0) {
+              numReplacingExecutors -= 1
+              logDebug(s"Decremented number of replaceing executors ($numReplacingExecutors left)")
+            }
           }
           // Note: some tests expect the reply to come after we put the executor in the map
           context.reply(RegisteredExecutor)
@@ -236,6 +246,12 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
             addressToExecutorId -= executorInfo.executorAddress
             executorDataMap -= executorId
             executorsPendingToRemove -= executorId
+            if (executorsToReplace.contains(executorId)) {
+              executorsToReplace -= executorId
+              if (numReplacingExecutors > 0) {
+                numReplacingExecutors -= 1
+              }
+            }
           }
           totalCoreCount.addAndGet(-executorInfo.totalCores)
           totalRegisteredExecutors.addAndGet(-1)
@@ -431,7 +447,11 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
     // take into account executors that are pending to be added or removed.
     if (!replace) {
       doRequestTotalExecutors(
-        numExistingExecutors + numPendingExecutors - executorsPendingToRemove.size)
+        numExistingExecutors + numPendingExecutors - executorsPendingToRemove.size
+        + numReplacingExecutors)
+    } else {
+      executorsToReplace ++= knownExecutors
+      numReplacingExecutors += knownExecutors.size
     }
 
     doKillExecutors(executorsToKill)