Make sure all required classes are present in the jar. Fix import order.

Author: harishreedharan

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

@@ -174,16 +174,20 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
       // Can't use Option[ExecutorDelegationTokenUpdater] because it is built only in YARN
       // profile, so use Option[Any] since even the stop method call will be via reflection.
       var tokenUpdaterOption: Option[Any] = None
-      val tokenUpdaterClass =
-        Class.forName("org.apache.spark.deploy.yarn.ExecutorDelegationTokenUpdater")
+      var tokenUpdaterClass: Option[Class[_]] = None
       if (driverConf.contains("spark.yarn.credentials.file")) {
         logInfo("Will periodically update credentials from: " +
           driverConf.get("spark.yarn.credentials.file"))
+
+        tokenUpdaterClass =
+          Some(Class.forName("org.apache.spark.deploy.yarn.ExecutorDelegationTokenUpdater"))
+
         // Periodically update the credentials for this user to ensure HDFS tokens get updated.
         val constructor  =
-          tokenUpdaterClass.getDeclaredConstructor(classOf[SparkConf], classOf[Configuration])
+          tokenUpdaterClass.get.getDeclaredConstructor(classOf[SparkConf], classOf[Configuration])
         tokenUpdaterOption = Some(constructor.newInstance(driverConf, SparkHadoopUtil.get.conf))
-        tokenUpdaterClass.getMethod("updateCredentialsIfRequired").invoke(tokenUpdaterOption.get)
+        tokenUpdaterClass.get.getMethod("updateCredentialsIfRequired")
+          .invoke(tokenUpdaterOption.get)
       }
 
       val env = SparkEnv.createExecutorEnv(
@@ -202,7 +206,7 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
       }
       env.rpcEnv.awaitTermination()
       if (tokenUpdaterOption.isDefined) {
-        tokenUpdaterClass.getMethod("stop").invoke(tokenUpdaterOption.get)
+        tokenUpdaterClass.get.getMethod("stop").invoke(tokenUpdaterOption.get)
       }
     }
   }

yarn/src/main/scala/org/apache/spark/deploy/yarn/ExecutorDelegationTokenUpdater.scala

@@ -21,9 +21,10 @@ import java.util.concurrent.{Executors, TimeUnit}
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.hadoop.security.{Credentials, UserGroupInformation}
+
 import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.util.{ThreadUtils, Utils}
 import org.apache.spark.{Logging, SparkConf}
+import org.apache.spark.util.{ThreadUtils, Utils}
 
 import scala.util.control.NonFatal