addressed comments

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

@@ -46,11 +46,10 @@ import org.apache.spark.util.{RpcUtils, SerializableBuffer, ThreadUtils, Utils}
  */
 private[spark]
 class CoarseGrainedSchedulerBackend(
-    scheduler: TaskSchedulerImpl,
-    val rpcEnv: RpcEnv,
-    hadoopDelegationTokenManager: Option[HadoopDelegationTokenManager])
-    extends ExecutorAllocationClient with SchedulerBackend with Logging
-{
+  scheduler: TaskSchedulerImpl,
+  val rpcEnv: RpcEnv)
+  extends ExecutorAllocationClient with SchedulerBackend with Logging {
+
   // Use an atomic variable to track total number of cores in the cluster for simplicity and speed
   protected val totalCoreCount = new AtomicInteger(0)
   // Total number of executors that are currently registered
@@ -102,6 +101,9 @@ class CoarseGrainedSchedulerBackend(
   // The num of current max ExecutorId used to re-register appMaster
   @volatile protected var currentExecutorIdCounter = 0
 
+  // hadoop token manager used by some sub-classes (e.g. Mesos)
+  protected val hadoopDelegationTokenManager: Option[HadoopDelegationTokenManager] = None
+
   // Hadoop delegation tokens to be sent to the executors.
   private val hadoopDelegationCreds = getHadoopDelegationCreds()
 
@@ -689,7 +691,7 @@ class CoarseGrainedSchedulerBackend(
   }
 
   private def getHadoopDelegationCreds(): Option[Array[Byte]] = {
-    if (UserGroupInformation.isSecurityEnabled) {
+    if (UserGroupInformation.isSecurityEnabled && hadoopDelegationTokenManager.isDefined) {
       hadoopDelegationTokenManager.map { manager =>
         val creds = UserGroupInformation.getCurrentUser.getCredentials
         val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)

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

@@ -38,7 +38,7 @@ private[spark] class StandaloneSchedulerBackend(
     scheduler: TaskSchedulerImpl,
     sc: SparkContext,
     masters: Array[String])
-  extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv, None)
+  extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv)
   with StandaloneAppClientListener
   with Logging {
 

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

@@ -56,13 +56,13 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
     securityManager: SecurityManager)
   extends CoarseGrainedSchedulerBackend(
     scheduler,
-    sc.env.rpcEnv,
-    Some(new HadoopDelegationTokenManager(
-      sc.conf,
-      sc.hadoopConfiguration)))
+    sc.env.rpcEnv)
     with org.apache.mesos.Scheduler
     with MesosSchedulerUtils {
 
+  override val hadoopDelegationTokenManager: Option[HadoopDelegationTokenManager] =
+    Some(new HadoopDelegationTokenManager(sc.conf, sc.hadoopConfiguration))
+
   // Blacklist a slave after this many failures
   private val MAX_SLAVE_FAILURES = 2
 

resource-managers/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala

@@ -40,7 +40,7 @@ import org.apache.spark.util.{RpcUtils, ThreadUtils}
 private[spark] abstract class YarnSchedulerBackend(
     scheduler: TaskSchedulerImpl,
     sc: SparkContext)
-  extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv, None) {
+  extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
 
   private val stopped = new AtomicBoolean(false)