WIP: Add kerberos principal and keytab to YARN client.

Author: harishreedharan

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

@@ -17,8 +17,10 @@
 
 package org.apache.spark.deploy.yarn
 
+import java.io.File
 import java.net.{InetAddress, UnknownHostException, URI, URISyntaxException}
 import java.nio.ByteBuffer
+import java.nio.file.Files
 
 import scala.collection.JavaConversions._
 import scala.collection.mutable.{ArrayBuffer, HashMap, ListBuffer, Map}
@@ -61,7 +63,7 @@ private[spark] class Client(
 
   private val yarnClient = YarnClient.createYarnClient
   private val yarnConf = new YarnConfiguration(hadoopConf)
-  private val credentials = UserGroupInformation.getCurrentUser.getCredentials
+  private var credentials: Credentials = null
   private val amMemoryOverhead = args.amMemoryOverhead // MB
   private val executorMemoryOverhead = args.executorMemoryOverhead // MB
   private val distCacheMgr = new ClientDistributedCacheManager()
@@ -540,6 +542,23 @@ private[spark] class Client(
     amContainer
   }
 
+  def setupCredentials(): Unit = {
+    Option(args.principal) match {
+      case Some(principal) =>
+        Option(args.keytab) match {
+          case Some(keytabPath) =>
+            File principalFile = Files.createTempF
+            val ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keytabPath)
+            credentials = ugi.getCredentials
+          case None =>
+        }
+      case None =>
+        credentials = UserGroupInformation.getCurrentUser.getCredentials
+
+    }
+
+  }
+
   /**
    * Report the state of an application until it has exited, either successfully or
    * due to some failure, then return a pair of the yarn application state (FINISHED, FAILED,

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

@@ -41,6 +41,8 @@ private[spark] class ClientArguments(args: Array[String], sparkConf: SparkConf)
   var amCores: Int = 1
   var appName: String = "Spark"
   var priority = 0
+  var principal: String = null
+  var keytab: String = null
   def isClusterMode: Boolean = userClass != null
 
   private var driverMemory: Int = 512 // MB
@@ -222,6 +224,14 @@ private[spark] class ClientArguments(args: Array[String], sparkConf: SparkConf)
           archives = value
           args = tail
 
+        case ("--principal") :: value :: tail =>
+          principal = value
+          args = tail
+
+        case ("--keytab") :: value :: tail =>
+          keytab = value
+          args = tail
+
         case Nil =>
 
         case _ =>