secure-token

diff --git examples/src/main/scala/org/apache/spark/examples/sql/hive/HiveTest.scala examples/src/main/scala/org/apache/spark/examples/sql/hive/HiveTest.scala
new file mode 100644
index 0000000..fb945ed
--- /dev/null
+++ examples/src/main/scala/org/apache/spark/examples/sql/hive/HiveTest.scala
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.examples.sql.hive
+
+import org.apache.spark.{SparkContext, SparkConf}
+import com.google.common.io.{ByteStreams, Files}
+
+import java.io.File
+
+import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.spark.sql._
+import org.apache.spark.sql.hive.HiveContext
+
+
+object HiveTest {
+  def main(args: Array[String]){
+    val appName = "HiveTest"
+    val conf = new SparkConf()
+    conf.setAppName(appName)
+    val sc = new SparkContext(conf)
+    val sqlContext = new HiveContext(sc)
+    val results = sqlContext.sql("show databases").collect()
+    println("Results.count" + results.length)
+    results.foreach(println)
+    println("----")
+    val table = sqlContext.sql("show tables").collect()
+    println("table.count" + table.length)
+    table.foreach(println)
+    println("----")
+    sc.stop()
+  }
+}
diff --git yarn/pom.xml yarn/pom.xml
index 5408062..4e5e4ad 100644
--- yarn/pom.xml
+++ yarn/pom.xml
@@ -97,6 +97,11 @@
       <artifactId>mockito-all</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>javassist</groupId>
+      <artifactId>javassist</artifactId>
+      <version>3.12.1.GA</version>
+    </dependency>
   </dependencies>

   <!--
   diff --git yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
index 61f8fc3..9d831c9 100644
--- yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
+++ yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
@@ -22,17 +22,21 @@ import java.nio.ByteBuffer

 import scala.collection.JavaConversions._
 import scala.collection.mutable.{ArrayBuffer, HashMap, ListBuffer, Map}
+import scala.reflect.runtime.universe
 import scala.util.{Try, Success, Failure}

 import com.google.common.base.Objects

 import org.apache.hadoop.io.DataOutputBuffer
 import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier
 import org.apache.hadoop.fs._
 import org.apache.hadoop.fs.permission.FsPermission
+import org.apache.hadoop.io.Text
 import org.apache.hadoop.mapred.Master
 import org.apache.hadoop.mapreduce.MRJobConfig
 import org.apache.hadoop.security.{Credentials, UserGroupInformation}
+import org.apache.hadoop.security.token.Token
 import org.apache.hadoop.util.StringUtils
 import org.apache.hadoop.yarn.api._
 import org.apache.hadoop.yarn.api.ApplicationConstants.Environment
@@ -217,6 +221,7 @@ private[spark] class Client(
     val dst = new Path(fs.getHomeDirectory(), appStagingDir)
     val nns = getNameNodesToAccess(sparkConf) + dst
     obtainTokensForNamenodes(nns, hadoopConf, credentials)
+    obtainTokenForHiveMetastore(hadoopConf, credentials)

     val replication = sparkConf.getInt("spark.yarn.submit.file.replication",
       fs.getDefaultReplication(dst)).toShort
@@ -902,6 +907,67 @@ object Client extends Logging {
    }
   }

+
+  /**
+   * Obtains token for the Hive metastore and adds them to the credentials.
+   */
+  private def obtainTokenForHiveMetastore(conf: Configuration, credentials: Credentials) {
+    org.apache.spark.deploy.yarn.HiveInstrumentationAgentYarn.instrument
+    if (UserGroupInformation.isSecurityEnabled) {
+      val mirror = universe.runtimeMirror(getClass.getClassLoader)
+
+      try {
+        val hiveClass = mirror.classLoader.loadClass("org.apache.hadoop.hive.ql.metadata.Hive")
+        val hive = hiveClass.getMethod("get").invoke(null)
+
+        val hiveConf = hiveClass.getMethod("getConf").invoke(hive)
+        val hiveConfClass = mirror.classLoader.loadClass("org.apache.hadoop.hive.conf.HiveConf")
+
+        val hiveConfGet = (param:String) => Option(hiveConfClass
+          .getMethod("get", classOf[java.lang.String])
+          .invoke(hiveConf, param))
+
+        val metastore_uri = hiveConfGet("hive.metastore.uris")
+
+        // Check for local metastore
+        if (metastore_uri != None && metastore_uri.get.toString.size > 0) {
+          val metastore_kerberos_principal_conf_var = mirror.classLoader
+            .loadClass("org.apache.hadoop.hive.conf.HiveConf$ConfVars")
+            .getField("METASTORE_KERBEROS_PRINCIPAL").get("varname").toString
+
+          val principal = hiveConfGet(metastore_kerberos_principal_conf_var)
+
+          val username = Option(UserGroupInformation.getCurrentUser().getUserName)
+          if (principal != None && username != None) {
+            val tokenStr = hiveClass.getMethod("getDelegationToken",
+              classOf[java.lang.String], classOf[java.lang.String])
+              .invoke(hive, username.get, username.get).asInstanceOf[java.lang.String]
+
+            val hive2Token = new Token[DelegationTokenIdentifier]()
+            hive2Token.decodeFromUrlString(tokenStr)
+            //config.set("hive.metastore.token.signature", "hive.server2.delegation.token")
+            credentials.addToken(new Text("hive.server2.delegation.token"),hive2Token)
+            logDebug("Added hive.Server2.delegation.token to conf.")
+            hiveClass.getMethod("closeCurrent").invoke(null)
+          } else {
+            logError("Username or principal == NULL")
+            logError(s"""username=${username.getOrElse("(NULL)")}""")
+            logError(s"""principal=${principal.getOrElse("(NULL)")}""")
+            throw new IllegalArgumentException("username and/or principal is equal to null!")
+          }
+        } else {
+          logDebug("HiveMetaStore configured in localmode")
+        }
+      } catch {
+        case e:java.lang.NoSuchMethodException => { logInfo("Hive Method not found " + e); return }
+        case e:java.lang.ClassNotFoundException => { logInfo("Hive Class not found " + e); return }
+        case e:Exception => { logError("Unexpected Exception " + e)
+          throw new RuntimeException("Unexpected exception", e)
+        }
+      }
+    }
+  }
+
   /**
    * Return whether the two file systems are the same.
    */
diff --git yarn/src/main/scala/org/apache/spark/deploy/yarn/HiveInstrumentationAgent.scala yarn/src/main/scala/org/apache/spark/deploy/yarn/HiveInstrumentationAgent.scala
new file mode 100644
index 0000000..6b5855e
--- /dev/null
+++ yarn/src/main/scala/org/apache/spark/deploy/yarn/HiveInstrumentationAgent.scala
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.yarn
+
+import javassist.ClassPool
+import javassist.CtClass
+import javassist.CtMethod
+import javassist.ClassMap
+import javassist.Modifier
+import javassist.CtField
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.HashMap
+import sun.misc.Unsafe
+
+object HiveInstrumentationAgentYarn {
+  var latch = new AtomicBoolean(false)
+  val unsafe = {
+    val field = classOf[Unsafe].getDeclaredField("theUnsafe");
+    field.setAccessible(true);
+    field.get(null).asInstanceOf[Unsafe]
+  }
+
+  private val pool = ClassPool.getDefault();
+  private val newClass = pool.get("org.apache.hadoop.hive.thrift.HadoopThriftAuthBridge23")
+  private val oldClass = pool.get("org.apache.hadoop.hive.thrift.HadoopThriftAuthBridge20S")
+
+  def instrument = {
+      if (!latch.getAndSet(true)) {
+        val targetMethods = oldClass.getDeclaredMethods
+        for (targetMethod <- targetMethods) {
+          if (targetMethod.getName() == "getHadoopSaslProperties") {
+            this.swapMethodBody(targetMethod)
+          }
+        }
+        val scBytes = this.oldClass.toBytecode()
+        unsafe.defineClass(null, scBytes, 0, scBytes.length,
+          this.getClass.getClassLoader(), this.getClass.getProtectionDomain())
+      }
+  }
+
+  private def swapMethodBody(targetMethod: CtMethod) {
+    val desc = targetMethod.getMethodInfo().getDescriptor()
+    try {
+      val sourceMethod = newClass.getMethod(targetMethod.getName(), desc)
+      targetMethod.setBody(sourceMethod, null)
+      println("swapMethodBody " + sourceMethod + " target: " + targetMethod)
+    } catch {
+      case _: Throwable =>
+    }
+
+  }
+}