[SPARK-13001][CORE][MESOS] Prevent getting offers in coarse-grained mode when reached max cores

Author: sebastienrainville

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala

@@ -109,10 +109,14 @@ private[spark] class CoarseMesosSchedulerBackend(
   private val slaveOfferConstraints =
     parseConstraintString(sc.conf.get("spark.mesos.constraints", ""))
 
-  // reject offers with mismatched constraints in seconds
+  // Reject offers with mismatched constraints in seconds
   private val rejectOfferDurationForUnmetConstraints =
     getRejectOfferDurationForUnmetConstraints(sc)
 
+  // Reject offers when we reached the maximum number of cores for this framework
+  private val rejectOfferDurationForReachedMaxCores =
+    getRejectOfferDurationForReachedMaxCores(sc)
+
   // A client for talking to the external shuffle service
   private val mesosExternalShuffleClient: Option[MesosExternalShuffleClient] = {
     if (shuffleServiceEnabled) {
@@ -273,25 +277,25 @@ private[spark] class CoarseMesosSchedulerBackend(
         matchesAttributeRequirements(slaveOfferConstraints, offerAttributes)
       }
 
-      declineUnmatchedOffers(d, unmatchedOffers)
+      unmatchedOffers.foreach { offer =>
+        declineOffer(d, offer, "unmet constraints", rejectOfferDurationForUnmetConstraints)
+      }
+
       handleMatchedOffers(d, matchedOffers)
     }
   }
 
-  private def declineUnmatchedOffers(d: SchedulerDriver, offers: Buffer[Offer]): Unit = {
-    for (offer <- offers) {
-      val id = offer.getId.getValue
-      val offerAttributes = toAttributeMap(offer.getAttributesList)
-      val mem = getResource(offer.getResourcesList, "mem")
-      val cpus = getResource(offer.getResourcesList, "cpus")
-      val filters = Filters.newBuilder()
-        .setRefuseSeconds(rejectOfferDurationForUnmetConstraints).build()
+  private def declineOffer(d: SchedulerDriver, offer: Offer, reason: String, refuseSeconds: Long) {
+    val id = offer.getId.getValue
+    val offerAttributes = toAttributeMap(offer.getAttributesList)
+    val mem = getResource(offer.getResourcesList, "mem")
+    val cpus = getResource(offer.getResourcesList, "cpus")
 
-      logDebug(s"Declining offer: $id with attributes: $offerAttributes mem: $mem cpu: $cpus"
-        + s" for $rejectOfferDurationForUnmetConstraints seconds")
+    logDebug(s"Declining offer ($reason): $id with attributes: $offerAttributes mem: $mem"
+      + s" cpu: $cpus for $rejectOfferDurationForUnmetConstraints seconds")
 
-      d.declineOffer(offer.getId, filters)
-    }
+    val filters = Filters.newBuilder().setRefuseSeconds(refuseSeconds).build()
+    d.declineOffer(offer.getId, filters)
   }
 
   /**
@@ -326,7 +330,9 @@ private[spark] class CoarseMesosSchedulerBackend(
         d.launchTasks(
           Collections.singleton(offer.getId),
           offerTasks.asJava)
-      } else { // decline
+      } else if (totalCoresAcquired >= maxCores) {
+        declineOffer(d, offer, "reached max cores", rejectOfferDurationForReachedMaxCores)
+      } else {
         logDebug(s"Declining offer: $id with attributes: $offerAttributes " +
           s"mem: $offerMem cpu: $offerCpus")
 

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala

@@ -352,4 +352,8 @@ private[mesos] trait MesosSchedulerUtils extends Logging {
     sc.conf.getTimeAsSeconds("spark.mesos.rejectOfferDurationForUnmetConstraints", "120s")
   }
 
+  protected def getRejectOfferDurationForReachedMaxCores(sc: SparkContext): Long = {
+    sc.conf.getTimeAsSeconds("spark.mesos.rejectOfferDurationForReachedMaxCores", "120s")
+  }
+
 }