working through comments

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

@@ -142,6 +142,17 @@ class SparkHadoopUtil extends Logging {
     }
   }
 
+  def addDelegationTokens(tokens: Array[Byte], sparkConf: SparkConf) {
+    logInfo(s"Found delegation tokens of ${tokens.length} bytes")
+    val hadoopConf = newConfiguration(sparkConf)
+    hadoopConf.set("hadoop.security.authentication", "Token")
+    UserGroupInformation.setConfiguration(hadoopConf)
+    // decode tokens and add them to the credentials
+    val creds = deserialize(tokens)
+    uglyF(s"creds $creds id")
+    addCurrentUserCredentials(deserialize(tokens))
+  }
+
   /**
    * Returns a function that can be called to find Hadoop FileSystem bytes read. If
    * getFSBytesReadOnThreadCallback is called from thread r at time t, the returned callback will

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

@@ -125,8 +125,11 @@ private[spark] class CoarseGrainedExecutorBackend(
           executor.stop()
         }
       }.start()
+
+    // This message is only sent by Mesos Drivers, and is not expected from other
+    // SchedulerBackends at this time
     case UpdateDelegationTokens(tokens) =>
-      CoarseGrainedExecutorBackend.addDelegationTokens(tokens, env.conf)
+      SparkHadoopUtil.get.addDelegationTokens(tokens, env.conf)
   }
 
   override def onDisconnected(remoteAddress: RpcAddress): Unit = {
@@ -178,16 +181,6 @@ private[spark] class CoarseGrainedExecutorBackend(
 
 private[spark] object CoarseGrainedExecutorBackend extends Logging {
 
-  private def addDelegationTokens(tokens: Array[Byte], sparkConf: SparkConf) {
-    logInfo(s"Found delegation tokens of ${tokens.length} bytes")
-    val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
-    hadoopConf.set("hadoop.security.authentication", "Token")
-    UserGroupInformation.setConfiguration(hadoopConf)
-    val creds = SparkHadoopUtil.get.deserialize(tokens)
-    // decode tokens and add them to the credentials
-    UserGroupInformation.getCurrentUser.addCredentials(SparkHadoopUtil.get.deserialize(tokens))
-  }
-
   private def run(
       driverUrl: String,
       executorId: String,

core/src/main/scala/org/apache/spark/internal/Logging.scala

@@ -54,6 +54,10 @@ trait Logging {
     if (log.isInfoEnabled) log.info(msg)
   }
 
+  protected def uglyF(msg: => String) {
+    if (log.isInfoEnabled) log.info("UGLY: " + msg)
+  }
+
   protected def logDebug(msg: => String) {
     if (log.isDebugEnabled) log.debug(msg)
   }

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

@@ -165,7 +165,6 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
         for ((x, executorData) <- executorDataMap) {
           executorData.executorEndpoint.send(UpdateDelegationTokens(tokens))
         }
-
     }
 
     override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {

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

@@ -198,16 +198,19 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
       sc.conf.getOption("spark.mesos.driver.frameworkId").map(_ + suffix)
     )
 
-    if (principal != null) {
+    // check that the credentials are defined, even though it's likely that auth would have failed
+    // adready if you've made it this far
+    if (principal != null && hadoopDelegationCreds.isDefined) {
       logDebug(s"Principal found ($principal) starting token renewer")
       val credentialRenewerThread = new Thread {
         setName("MesosCredentialRenewer")
         override def run(): Unit = {
+          val dummy: Option[Array[Byte]] = None
           val credentialRenewer =
             new MesosCredentialRenewer(
               conf,
               hadoopDelegationTokenManager.get,
-              MesosCredentialRenewer.getTokenRenewalInterval(hadoopDelegationCreds.get, conf),
+              MesosCredentialRenewer.getTokenRenewalTime(hadoopDelegationCreds.get, conf),
               driverEndpoint)
           credentialRenewer.scheduleTokenRenewal()
         }

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

@@ -63,7 +63,8 @@ class MesosCredentialRenewer(
 
   def scheduleTokenRenewal(): Unit = {
     def scheduleRenewal(runnable: Runnable): Unit = {
-      val remainingTime = timeOfNextRenewal - System.currentTimeMillis()
+      // val remainingTime = timeOfNextRenewal - System.currentTimeMillis()
+      val remainingTime = 5000
       if (remainingTime <= 0) {
         logInfo("Credentials have expired, creating new ones now.")
         runnable.run()
@@ -136,15 +137,15 @@ class MesosCredentialRenewer(
 }
 
 object MesosCredentialRenewer extends Logging {
-  def getTokenRenewalInterval(bytes: Array[Byte], conf: SparkConf): Long = {
+  def getTokenRenewalTime(bytes: Array[Byte], conf: SparkConf): Long = {
     val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)
     val creds = SparkHadoopUtil.get.deserialize(bytes)
-    val intervals = creds.getAllTokens.asScala.flatMap { t =>
+    val renewalTimes = creds.getAllTokens.asScala.flatMap { t =>
       Try {
         t.renew(hadoopConf)
       }.toOption
     }
-    if (intervals.isEmpty) Long.MaxValue else intervals.min
+    if (renewalTimes.isEmpty) Long.MaxValue else renewalTimes.min
   }
 }