Avoid assigning tasks to lost executors

markhamstra · Jul 20, 2015 · 30a9ad0 · 30a9ad0
1 parent 5bdf16d
commit 30a9ad0
Showing 1 changed file with 9 additions and 5 deletions.
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
@@ -169,8 +169,10 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
     // Make fake resource offers on all executors
     private def makeOffers() {
-      launchTasks(scheduler.resourceOffers(executorDataMap.map { case (id, executorData) =>
-        new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
+      launchTasks(scheduler.resourceOffers(executorDataMap
+        .filterKeys(!executorsPendingToRemove.contains(_))
+        .map { case (id, executorData) =>
+          new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
       }.toSeq))
     }
 
@@ -181,9 +183,11 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
     // Make fake resource offers on just one executor
     private def makeOffers(executorId: String) {
-      val executorData = executorDataMap(executorId)
-      launchTasks(scheduler.resourceOffers(
-        Seq(new WorkerOffer(executorId, executorData.executorHost, executorData.freeCores))))
+      if (!executorsPendingToRemove.contains(executorId)) {
+        val executorData = executorDataMap(executorId)
+        launchTasks(scheduler.resourceOffers(
+          Seq(new WorkerOffer(executorId, executorData.executorHost, executorData.freeCores))))
+      }
     }
 
     // Launch tasks returned by a set of resource offers