Fixed impersonation when data read from MapR-DB via Spark-Hive. (apache#131)

Author: Mikhail Gorbov

core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.rdd
 
 import java.io.{FileNotFoundException, IOException}
+import java.security.PrivilegedExceptionAction
 import java.text.SimpleDateFormat
 import java.util.{Date, Locale}
 
@@ -29,6 +30,7 @@ import org.apache.hadoop.mapred._
 import org.apache.hadoop.mapred.lib.CombineFileSplit
 import org.apache.hadoop.mapreduce.TaskType
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
+import org.apache.hadoop.security.UserGroupInformation
 import org.apache.hadoop.util.ReflectionUtils
 
 import org.apache.spark._
@@ -124,6 +126,8 @@ class HadoopRDD[K, V](
       minPartitions)
   }
 
+  private val doAsUserName = UserGroupInformation.getCurrentUser.getUserName
+
   protected val jobConfCacheKey: String = "rdd_%d_job_conf".format(id)
 
   protected val inputFormatCacheKey: String = "rdd_%d_input_format".format(id)
@@ -220,7 +224,7 @@ class HadoopRDD[K, V](
     }
   }
 
-  override def compute(theSplit: Partition, context: TaskContext): InterruptibleIterator[(K, V)] = {
+  def doCompute(theSplit: Partition, context: TaskContext): InterruptibleIterator[(K, V)] = {
     val iter = new NextIterator[(K, V)] {
 
       private val split = theSplit.asInstanceOf[HadoopPartition]
@@ -326,7 +330,7 @@ class HadoopRDD[K, V](
           if (getBytesReadCallback.isDefined) {
             updateBytesRead()
           } else if (split.inputSplit.value.isInstanceOf[FileSplit] ||
-                     split.inputSplit.value.isInstanceOf[CombineFileSplit]) {
+            split.inputSplit.value.isInstanceOf[CombineFileSplit]) {
             // If we can't get the bytes read from the FS stats, fall back to the split size,
             // which may be inaccurate.
             try {
@@ -342,6 +346,29 @@ class HadoopRDD[K, V](
     new InterruptibleIterator[(K, V)](context, iter)
   }
 
+  override def compute(theSplit: Partition, context: TaskContext): InterruptibleIterator[(K, V)] = {
+    val ugi = UserGroupInformation.getCurrentUser
+
+    if (ugi.getUserName == doAsUserName) {
+      doCompute(theSplit: Partition, context: TaskContext)
+    } else {
+      val doAsAction = new PrivilegedExceptionAction[InterruptibleIterator[(K, V)]]() {
+        override def run(): InterruptibleIterator[(K, V)] = {
+          try {
+            doCompute(theSplit: Partition, context: TaskContext)
+          } catch {
+            case e: Exception =>
+              log.error("Error when HadoopRDD computing: ", e)
+              throw e
+          }
+        }
+      }
+
+      val proxyUgi = UserGroupInformation.createProxyUser(doAsUserName, ugi)
+      proxyUgi.doAs(doAsAction)
+    }
+  }
+
   /** Maps over a partition, providing the InputSplit that was used as the base of the partition. */
   @DeveloperApi
   def mapPartitionsWithInputSplit[U: ClassTag](