From fd1f73531514865ecf0b632af628650b0b6f1983 Mon Sep 17 00:00:00 2001
From: huangzhaowei <huangzhaowei@huawei.com>
Date: Wed, 30 Sep 2015 10:03:00 +0800
Subject: [PATCH] [SPARK-10473][YARN]Login again in the driver to avoid the
 events lossing.

---
 .../scala/org/apache/spark/SparkContext.scala    | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 0c72adfb9505b..4861140a862c4 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -35,6 +35,7 @@ import scala.util.control.NonFatal
 
 import org.apache.commons.lang.SerializationUtils
 import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.security.UserGroupInformation
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.io.{ArrayWritable, BooleanWritable, BytesWritable, DoubleWritable,
   FloatWritable, IntWritable, LongWritable, NullWritable, Text, Writable}
@@ -440,6 +441,16 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
 
     _conf.set("spark.externalBlockStore.folderName", externalBlockStoreFolderName)
 
+    var principal: String = null
+    var keytab: String = null
+
+    if (master.startsWith("yarn")) {
+      if (_conf.contains("spark.yarn.principal") && _conf.contains("spark.yarn.keytab")) {
+        principal = _conf.get("spark.yarn.principal")
+        keytab = _conf.get("spark.yarn.keytab")
+      }
+    }
+
     if (master == "yarn-client") System.setProperty("SPARK_YARN_MODE", "true")
 
     // "_jobProgressListener" should be set up before creating SparkEnv because when creating
@@ -569,6 +580,11 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
       }
     _cleaner.foreach(_.start())
 
+    if (null != principal && UserGroupInformation.isSecurityEnabled) {
+      UserGroupInformation.loginUserFromKeytab(principal, keytab)
+      logInfo("Successfully logged into the KDC in SparkContext.")
+    }
+
     setupAndStartListenerBus()
     postEnvironmentUpdate()
     postApplicationStart()