Restore original RegisteredExecutor message, and send new tokens via NewTokens message.

Author: harishreedharan

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

@@ -72,17 +72,17 @@ private[spark] class CoarseGrainedExecutorBackend(
   }
 
   override def receive: PartialFunction[Any, Unit] = {
-    case RegisteredExecutor(tokens) =>
+    case RegisteredExecutor =>
       logInfo("Successfully registered with driver")
       val (hostname, _) = Utils.parseHostPort(hostPort)
       executor = new Executor(executorId, hostname, env, userClassPath, isLocal = false)
-      tokens.foreach { x =>
-        val inStream = new DataInputStream(new ByteArrayInputStream(x.value.array()))
-        val creds = new Credentials()
-        creds.readFields(inStream)
-        inStream.close()
-        UserGroupInformation.getCurrentUser.addCredentials(creds)
-      }
+
+    case NewTokens(tokens) =>
+      val inStream = new DataInputStream(new ByteArrayInputStream(tokens.value.array()))
+      val creds = new Credentials()
+      creds.readFields(inStream)
+      inStream.close()
+      UserGroupInformation.getCurrentUser.addCredentials(creds)
 
     case RegisterExecutorFailed(message) =>
       logError("Slave registration failed: " + message)

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

@@ -35,8 +35,7 @@ private[spark] object CoarseGrainedClusterMessages {
   case class KillTask(taskId: Long, executor: String, interruptThread: Boolean)
     extends CoarseGrainedClusterMessage
 
-  case class RegisteredExecutor(tokens: Option[SerializableBuffer])
-    extends CoarseGrainedClusterMessage
+  case object RegisteredExecutor extends CoarseGrainedClusterMessage
 
   case class NewTokens(tokens: SerializableBuffer) extends CoarseGrainedClusterMessage
 

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

@@ -124,8 +124,8 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
           context.reply(RegisterExecutorFailed("Duplicate executor ID: " + executorId))
         } else {
           logInfo("Registered executor: " + executorRef + " with ID " + executorId)
-          context.reply(RegisteredExecutor(latestTokens))
-
+          context.reply(RegisteredExecutor)
+          latestTokens.foreach(x => context.reply(NewTokens(x)))
           addressToExecutorId(executorRef.address) = executorId
           totalCoreCount.addAndGet(cores)
           totalRegisteredExecutors.addAndGet(1)