SPARK-19755 declining offers from blacklisted slave by BlacklistTracker

Author: IgorBerman

resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackend.scala

@@ -568,6 +568,7 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
       cpus + totalCoresAcquired <= maxCores &&
       mem <= offerMem &&
       numExecutors < executorLimit &&
+      !scheduler.nodeBlacklist().contains(offerHostname) &&
       meetsPortRequirements &&
       satisfiesLocality(offerHostname)
   }
@@ -792,4 +793,4 @@ object IdHelper {
   // Use atomic values since Spark contexts can be initialized in parallel
   private[mesos] val nextSCNumber = new AtomicLong(0)
   private[mesos] val startedBefore = new AtomicBoolean(false)
-}
+}

resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackendSuite.scala

@@ -108,6 +108,28 @@ class MesosCoarseGrainedSchedulerBackendSuite extends SparkFunSuite
     verifyTaskLaunched(driver, "o2")
   }
 
+  test("mesos declines offers from blacklisted slave") {
+    setBackend()
+
+    // launches a task on a valid offer on slave s1
+    val minMem = backend.executorMemory(sc) + 1024
+    val minCpu = 4
+    val offer1 = Resources(minMem, minCpu)
+    offerResources(List(offer1))
+    verifyTaskLaunched(driver, "o1")
+
+    // for any reason executor(aka mesos task) failed on s1
+    val status = createTaskStatus("0", "s1", TaskState.TASK_FAILED)
+    backend.statusUpdate(driver, status)
+    when(taskScheduler.nodeBlacklist()).thenReturn(Set("hosts1"))
+
+    val offer2 = Resources(minMem, minCpu)
+    // Offer resources from the same slave
+    offerResources(List(offer2))
+    // but since it's blacklisted the offer is declined
+    verifyDeclinedOffer(driver, createOfferId("o1"))
+  }
+
   test("mesos supports spark.executor.cores") {
     val executorCores = 4
     setBackend(Map("spark.executor.cores" -> executorCores.toString))
@@ -790,6 +812,7 @@ class MesosCoarseGrainedSchedulerBackendSuite extends SparkFunSuite
 
     taskScheduler = mock[TaskSchedulerImpl]
     when(taskScheduler.sc).thenReturn(sc)
+    when(taskScheduler.nodeBlacklist()).thenReturn(Set[String]())
 
     externalShuffleClient = mock[MesosExternalShuffleClient]