[spark] branch branch-2.4 updated: [SPARK-26192][MESOS][2.4] Retrieve enableFetcherCache option from submission for driver URIs

[do...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dongjoon pushed a commit to branch branch-2.4
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-2.4 by this push:
     new 2adf548  [SPARK-26192][MESOS][2.4] Retrieve enableFetcherCache option from submission for driver URIs
2adf548 is described below

commit 2adf548c26cd3e2dd96bad5d33f2d212fa2c99ba
Author: mwlon <ml...@hmc.edu>
AuthorDate: Fri May 31 08:09:47 2019 -0700

    [SPARK-26192][MESOS][2.4] Retrieve enableFetcherCache option from submission for driver URIs
    
    ## What changes were proposed in this pull request?
    
    Retrieve enableFetcherCache option from submission conf rather than dispatcher conf. This resolves some confusing behavior where Spark drivers currently get this conf from the dispatcher, whereas Spark executors get this conf from the submission. After this change, the conf will only need to be specified once.
    
    ## How was this patch tested?
    
    With (updated) existing tests.
    
    Closes #24750 from mwlon/SPARK-26192-min-2.4.
    
    Authored-by: mwlon <ml...@hmc.edu>
    Signed-off-by: Dongjoon Hyun <dh...@apple.com>
---
 .../cluster/mesos/MesosClusterScheduler.scala      |  4 ++-
 .../cluster/mesos/MesosClusterSchedulerSuite.scala | 29 ++++++++++++++++++++--
 2 files changed, 30 insertions(+), 3 deletions(-)

diff --git a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
index 7a23862..9dbe9fd 100644
--- a/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
+++ b/resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala
@@ -129,7 +129,6 @@ private[spark] class MesosClusterScheduler(
   private val queuedCapacity = conf.getInt("spark.mesos.maxDrivers", 200)
   private val retainedDrivers = conf.getInt("spark.mesos.retainedDrivers", 200)
   private val maxRetryWaitTime = conf.getInt("spark.mesos.cluster.retry.wait.max", 60) // 1 minute
-  private val useFetchCache = conf.getBoolean("spark.mesos.fetcherCache.enable", false)
   private val schedulerState = engineFactory.createEngine("scheduler")
   private val stateLock = new Object()
   // Keyed by submission id
@@ -430,6 +429,9 @@ private[spark] class MesosClusterScheduler(
   }
 
   private def getDriverUris(desc: MesosDriverDescription): List[CommandInfo.URI] = {
+    val useFetchCache = desc.conf.getBoolean("spark.mesos.fetcherCache.enable", false) ||
+        conf.getBoolean("spark.mesos.fetcherCache.enable", false)
+
     val confUris = List(conf.getOption("spark.mesos.uris"),
       desc.conf.getOption("spark.mesos.uris"),
       desc.conf.getOption("spark.submit.pyFiles")).flatMap(
diff --git a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterSchedulerSuite.scala b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterSchedulerSuite.scala
index 11c48b5..ae001e2 100644
--- a/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterSchedulerSuite.scala
+++ b/resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterSchedulerSuite.scala
@@ -254,7 +254,7 @@ class MesosClusterSchedulerSuite extends SparkFunSuite with LocalSparkContext wi
     assert(networkInfos.get(0).getLabels.getLabels(1).getValue == "val2")
   }
 
-  test("SPARK-26082 supports setting fetcher cache") {
+  test("supports setting fetcher cache on the dispatcher") {
     setScheduler(Map("spark.mesos.fetcherCache.enable" -> "true"))
 
     val mem = 1000
@@ -278,7 +278,32 @@ class MesosClusterSchedulerSuite extends SparkFunSuite with LocalSparkContext wi
     assert(uris.asScala.forall(_.getCache))
   }
 
-  test("SPARK-26082 supports disabling fetcher cache") {
+  test("supports setting fetcher cache in the submission") {
+    setScheduler(Map())
+
+    val mem = 1000
+    val cpu = 1
+
+    val response = scheduler.submitDriver(
+      new MesosDriverDescription("d1", "jar", mem, cpu, true,
+        command,
+        Map("spark.mesos.executor.home" -> "test",
+          "spark.app.name" -> "test",
+          "spark.mesos.fetcherCache.enable" -> "true"),
+        "s1",
+        new Date()))
+
+    assert(response.success)
+
+    val offer = Utils.createOffer("o1", "s1", mem, cpu)
+    scheduler.resourceOffers(driver, List(offer).asJava)
+
+    val launchedTasks = Utils.verifyTaskLaunched(driver, "o1")
+    val uris = launchedTasks.head.getCommand.getUrisList
+    assert(uris.asScala.forall(_.getCache))
+  }
+
+  test("supports disabling fetcher cache") {
     setScheduler(Map("spark.mesos.fetcherCache.enable" -> "false"))
 
     val mem = 1000


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org