[SPARK-19606] Support constraints in spark-dispatcher; updates to docs and tests

Author: Paul Mackles

docs/running-on-mesos.md

@@ -464,7 +464,8 @@ See the [configuration page](configuration.html) for information on Spark config
   <td><code>spark.mesos.driver.constraints</code></td>
   <td>(none)</td>
   <td>
-    Same as <code>spark.mesos.constraints</code> except applied to drivers.
+    Same as <code>spark.mesos.constraints</code> except applied to drivers when launched through the dispatcher. By default,
+    all resource offers will be accepted.
   </td>
 </tr>
 <tr>

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

@@ -254,30 +254,51 @@ class MesosClusterSchedulerSuite extends SparkFunSuite with LocalSparkContext wi
     assert(networkInfos.get(0).getLabels.getLabels(1).getValue == "val2")
   }
 
-  test("declines offers that violate driver constraints") {
+  test("accept/decline offers with driver constraints") {
     setScheduler()
 
     val mem = 1000
     val cpu = 1
-    val s2Attributes = List(Utils.createTextAttribute("c", "u"))
+    val s2Attributes = List(Utils.createTextAttribute("c1", "a"))
+    val s3Attributes = List(
+      Utils.createTextAttribute("c1", "a"),
+      Utils.createTextAttribute("c2", "b"))
+    val offers = List(
+      Utils.createOffer("o1", "s1", mem, cpu, None, 0),
+      Utils.createOffer("o2", "s2", mem, cpu, None, 0, s2Attributes),
+      Utils.createOffer("o3", "s3", mem, cpu, None, 0, s3Attributes))
+
+    def submitDriver(driverConstraints: String): Unit = {
+      val response = scheduler.submitDriver(
+        new MesosDriverDescription("d1", "jar", mem, cpu, true,
+          command,
+          Map("spark.mesos.executor.home" -> "test",
+            "spark.app.name" -> "test",
+            "spark.mesos.driver.constraints" -> driverConstraints),
+          "s1",
+          new Date()))
+      assert(response.success)
+    }
 
-    val response = scheduler.submitDriver(
-      new MesosDriverDescription("d1", "jar", mem, cpu, true,
-        command,
-        Map("spark.mesos.executor.home" -> "test",
-          "spark.app.name" -> "test",
-          "spark.mesos.driver.constraints" -> "c:v"),
-        "s1",
-        new Date()))
+    submitDriver("c1:x")
+    scheduler.resourceOffers(driver, offers.asJava)
+    offers.foreach(o => Utils.verifyTaskNotLaunched(driver, o.getId.getValue))
 
-    assert(response.success)
+    submitDriver("c1:y;c2:z")
+    scheduler.resourceOffers(driver, offers.asJava)
+    offers.foreach(o => Utils.verifyTaskNotLaunched(driver, o.getId.getValue))
+
+    submitDriver("")
+    scheduler.resourceOffers(driver, offers.asJava)
+    Utils.verifyTaskLaunched(driver, "o1")
 
-    val offer1 = Utils.createOffer("o1", "s1", mem, cpu, None, 0)
-    val offer2 = Utils.createOffer("o2", "s2", mem, cpu, None, 0, s2Attributes)
-    scheduler.resourceOffers(driver, List(offer1, offer2).asJava)
+    submitDriver("c1:a")
+    scheduler.resourceOffers(driver, offers.asJava)
+    Utils.verifyTaskLaunched(driver, "o2")
 
-    Utils.verifyTaskNotLaunched(driver, "o1")
-    Utils.verifyTaskNotLaunched(driver, "o2")
+    submitDriver("c1:a;c2:b")
+    scheduler.resourceOffers(driver, offers.asJava)
+    Utils.verifyTaskLaunched(driver, "o3")
   }
 
   test("supports spark.mesos.driver.labels") {