Fix zk conf key for mesos zk engine.

Author: tnachen

core/src/main/scala/org/apache/spark/deploy/SparkCuratorUtil.scala

@@ -32,8 +32,10 @@ private[spark] object SparkCuratorUtil extends Logging {
   private val RETRY_WAIT_MILLIS = 5000
   private val MAX_RECONNECT_ATTEMPTS = 3
 
-  def newClient(conf: SparkConf): CuratorFramework = {
-    val ZK_URL = conf.get("spark.deploy.zookeeper.url")
+  def newClient(
+      conf: SparkConf,
+      zkUrlConf: String = "spark.deploy.zookeeper.url"): CuratorFramework = {
+    val ZK_URL = conf.get(zkUrlConf)
     val zk = CuratorFrameworkFactory.newClient(ZK_URL,
       ZK_SESSION_TIMEOUT_MILLIS, ZK_CONNECTION_TIMEOUT_MILLIS,
       new ExponentialBackoffRetry(RETRY_WAIT_MILLIS, MAX_RECONNECT_ATTEMPTS))

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

@@ -55,7 +55,7 @@ private[spark] trait MesosClusterPersistenceEngine {
 private[spark] class ZookeeperMesosClusterPersistenceEngineFactory(conf: SparkConf)
   extends MesosClusterPersistenceEngineFactory(conf) {
 
-  lazy val zk = SparkCuratorUtil.newClient(conf)
+  lazy val zk = SparkCuratorUtil.newClient(conf, "spark.mesos.deploy.zookeeper.url")
 
   def createEngine(path: String): MesosClusterPersistenceEngine = {
     new ZookeeperMesosClusterPersistenceEngine(path, zk, conf)