Remove reflection, use a method in SparkHadoopUtil to update the token renewer.

Author: harishreedharan

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

@@ -292,6 +292,17 @@ class SparkHadoopUtil extends Logging {
       }
     }
   }
+
+  /**
+   * Start a thread to periodically update the current user's credentials with new delegation
+   * tokens so that writes to HDFS do not fail.
+   */
+  private[spark] def startExecutorDelegationTokenRenewer(conf: SparkConf){}
+
+  /**
+   * Stop the thread that does the delegation token updates.
+   */
+  private[spark] def stopExecutorDelegationTokenRenewer() {}
 }
 
 object SparkHadoopUtil {

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

@@ -170,24 +170,10 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
           driverConf.set(key, value)
         }
       }
-      // Delegation Token Updater is not supported in Hadoop 1, so use reflection.
-      // 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
-      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.get.getDeclaredConstructor(classOf[SparkConf], classOf[Configuration])
-        tokenUpdaterOption = Some(constructor.newInstance(driverConf, SparkHadoopUtil.get.conf))
-        tokenUpdaterClass.get.getMethod("updateCredentialsIfRequired")
-          .invoke(tokenUpdaterOption.get)
+        SparkHadoopUtil.get.startExecutorDelegationTokenRenewer(driverConf)
       }
 
       val env = SparkEnv.createExecutorEnv(
@@ -205,9 +191,7 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
         env.rpcEnv.setupEndpoint("WorkerWatcher", new WorkerWatcher(env.rpcEnv, url))
       }
       env.rpcEnv.awaitTermination()
-      if (tokenUpdaterOption.isDefined) {
-        tokenUpdaterClass.get.getMethod("stop").invoke(tokenUpdaterOption.get)
-      }
+      SparkHadoopUtil.get.stopExecutorDelegationTokenRenewer()
     }
   }
 

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

@@ -44,6 +44,8 @@ import org.apache.spark.util.Utils
  */
 class YarnSparkHadoopUtil extends SparkHadoopUtil {
 
+  private var tokenRenewer: Option[ExecutorDelegationTokenUpdater] = None
+
   override def transferCredentials(source: UserGroupInformation, dest: UserGroupInformation) {
     dest.addCredentials(source.getCredentials())
   }
@@ -125,6 +127,15 @@ class YarnSparkHadoopUtil extends SparkHadoopUtil {
     }
   }
 
+  private[spark] override def startExecutorDelegationTokenRenewer(sparkConf: SparkConf): Unit = {
+    tokenRenewer = Some(new ExecutorDelegationTokenUpdater(sparkConf, conf))
+    tokenRenewer.get.updateCredentialsIfRequired()
+  }
+
+  private[spark] override def stopExecutorDelegationTokenRenewer(): Unit ={
+    tokenRenewer.foreach(_.stop())
+  }
+
 }
 
 object YarnSparkHadoopUtil {