Fix application id to use correct cook application id for enabling ssl

Author: WenboZhao

cook/src/main/scala/org/apache/spark/scheduler/cluster/cook/CoarseCookSchedulerBackend.scala

@@ -129,7 +129,7 @@ class CoarseCookSchedulerBackend(
         if (!job.isSuccess && !isAborted) {
           totalFailures += 1
           logWarning(s"Job ${job.getUUID} has died. " +
-            s"Failure ($totalFailures/$schedulerConf.getMaximumExecutorFailures)")
+            s"Failure ($totalFailures/${schedulerConf.getMaximumExecutorFailures})")
           jobIds -= job.getUUID
           if (totalFailures >= schedulerConf.getMaximumExecutorFailures) {
             // TODO should we abort the outstanding tasks now?

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

@@ -216,6 +216,8 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
     val uri = conf.getOption("spark.executor.uri")
       .orElse(Option(System.getenv("SPARK_EXECUTOR_URI")))
 
+    val cookAppId = conf.get("spark.cook.applicationId", applicationId())
+
     if (uri.isEmpty) {
       val executorSparkHome = conf.getOption("spark.mesos.executor.home")
         .orElse(sc.getSparkHome())
@@ -230,7 +232,7 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
         s" --executor-id $taskId" +
         s" --hostname ${executorHostname(offer)}" +
         s" --cores $numCores" +
-        s" --app-id $appId")
+        s" --app-id $cookAppId")
     } else {
       // Grab everything to the first '.'. We'll use that and '*' to
       // glob the directory "correctly".
@@ -243,7 +245,8 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
         s" --hostname ${executorHostname(offer)}" +
         s" --cores $numCores" +
         s" --app-id $appId")
-      command.addUris(CommandInfo.URI.newBuilder().setValue(uri.get).setCache(useFetcherCache))
+        s" --app-id $cookAppId")
+      command.addUris(CommandInfo.URI.newBuilder().setValue(uri.get))
     }
 
     conf.getOption("spark.mesos.uris").foreach(setupUris(_, command, useFetcherCache))