Merge pull request #909 from mateiz/exec-id-fix

Fix an instance where full standalone mode executor IDs were passed to

core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterTaskSetManager.scala

@@ -335,7 +335,7 @@ private[spark] class ClusterTaskSetManager(
   }
 
   /**
-   * Respond to an offer of a single slave from the scheduler by finding a task
+   * Respond to an offer of a single executor from the scheduler by finding a task
    */
   override def resourceOffer(
       execId: String,
@@ -358,7 +358,7 @@ private[spark] class ClusterTaskSetManager(
           val task = tasks(index)
           val taskId = sched.newTaskId()
           // Figure out whether this should count as a preferred launch
-          logInfo("Starting task %s:%d as TID %s on slave %s: %s (%s)".format(
+          logInfo("Starting task %s:%d as TID %s on executor %s: %s (%s)".format(
             taskSet.id, index, taskId, execId, host, taskLocality))
           // Do various bookkeeping
           copiesRunning(index) += 1

core/src/main/scala/org/apache/spark/scheduler/cluster/SparkDeploySchedulerBackend.scala

@@ -76,17 +76,17 @@ private[spark] class SparkDeploySchedulerBackend(
     }
   }
 
-  override def executorAdded(executorId: String, workerId: String, hostPort: String, cores: Int, memory: Int) {
+  override def executorAdded(fullId: String, workerId: String, hostPort: String, cores: Int, memory: Int) {
     logInfo("Granted executor ID %s on hostPort %s with %d cores, %s RAM".format(
-       executorId, hostPort, cores, Utils.megabytesToString(memory)))
+      fullId, hostPort, cores, Utils.megabytesToString(memory)))
   }
 
-  override def executorRemoved(executorId: String, message: String, exitStatus: Option[Int]) {
+  override def executorRemoved(fullId: String, message: String, exitStatus: Option[Int]) {
     val reason: ExecutorLossReason = exitStatus match {
       case Some(code) => ExecutorExited(code)
       case None => SlaveLost(message)
     }
-    logInfo("Executor %s removed: %s".format(executorId, message))
-    removeExecutor(executorId, reason.toString)
+    logInfo("Executor %s removed: %s".format(fullId, message))
+    removeExecutor(fullId.split("/")(1), reason.toString)
   }
 }