[SPARK-23078] add tests for Spark Thrift Server on Kubernetes

dev/dev-run-integration-tests.sh

@@ -89,7 +89,7 @@ then
   fi
   cd $SPARK_REPO_LOCAL_DIR
   git checkout -B $BRANCH origin/$BRANCH
-  ./dev/make-distribution.sh --tgz -Phadoop-2.7 -Pkubernetes -DskipTests;
+  ./dev/make-distribution.sh --tgz -Phadoop-2.7 -Pkubernetes -Phive -Phive-thriftserver -DskipTests;
   SPARK_TGZ=$(find $SPARK_REPO_LOCAL_DIR -name spark-*.tgz)
   echo "Built Spark TGZ at $SPARK_TGZ".
   cd -

pom.xml

@@ -30,6 +30,9 @@
     <exec-maven-plugin.version>1.4.0</exec-maven-plugin.version>
     <extraScalaTestArgs></extraScalaTestArgs>
     <guava.version>18.0</guava.version>
+    <hadoop.version>2.6.5</hadoop.version>
+    <hive.group>org.spark-project.hive</hive.group>
+    <hive.version>1.2.1.spark2</hive.version>
     <jsr305.version>1.3.9</jsr305.version>
     <kubernetes-client.version>3.0.0</kubernetes-client.version>
     <log4j.version>1.2.17</log4j.version>
@@ -84,11 +87,21 @@
       <artifactId>log4j</artifactId>
       <version>${log4j.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-common</artifactId>
+      <version>${hadoop.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-lang3</artifactId>
       <version>${commons-lang3.version}</version>
     </dependency>
+    <dependency>
+      <groupId>${hive.group}</groupId>
+      <artifactId>hive-jdbc</artifactId>
+      <version>${hive.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.scala-lang</groupId>
       <artifactId>scala-library</artifactId>

src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala

@@ -20,6 +20,8 @@ import java.io.File
 import java.nio.file.{Path, Paths}
 import java.util.UUID
 import java.util.regex.Pattern
+import java.sql.DriverManager
+import org.apache.hive.jdbc.HiveDriver
 
 import scala.collection.JavaConverters._
 import com.google.common.io.PatternFilenameFilter
@@ -121,6 +123,10 @@ private[spark] class KubernetesSuite extends FunSuite with BeforeAndAfterAll wit
     runSparkPiAndVerifyCompletion(appArgs = Array("5"))
   }
 
+  test("Run Spark Thrift Server") {
+    runThriftServerAndVerifyQuery()
+  }
+
   test("Run SparkPi with custom labels, annotations, and environment variables.") {
     sparkAppConf
       .set("spark.kubernetes.driver.label.label1", "label1-value")
@@ -239,6 +245,43 @@ private[spark] class KubernetesSuite extends FunSuite with BeforeAndAfterAll wit
       appLocator)
   }
 
+  private def runThriftServerAndVerifyQuery(
+                                             driverPodChecker: Pod => Unit = doBasicDriverPodCheck,
+                                             appArgs: Array[String] = Array.empty[String],
+                                              appLocator: String = appLocator): Unit = {
+    val appArguments = SparkAppArguments(
+      mainAppResource = "",
+      mainClass = "org.apache.spark.sql.hive.thriftserver.HiveThriftServer2",
+      appArgs = appArgs)
+    SparkAppLauncher.launch(appArguments, sparkAppConf, TIMEOUT.value.toSeconds.toInt, sparkHomeDir)
+    val driverPod = kubernetesTestComponents.kubernetesClient
+      .pods
+      .withLabel("spark-app-locator", appLocator)
+      .withLabel("spark-role", "driver")
+      .list()
+      .getItems
+      .get(0)
+    driverPodChecker(driverPod)
+    val driverPodResource = kubernetesTestComponents.kubernetesClient
+      .pods
+      .withName(driverPod.getMetadata.getName)
+
+    Eventually.eventually(TIMEOUT, INTERVAL) {
+      val localPort = driverPodResource.portForward(10000).getLocalPort
+      val jdbcUri = s"jdbc:hive2://localhost:$localPort/"
+      val connection = DriverManager.getConnection(jdbcUri, "user", "pass")
+      val statement = connection.createStatement()
+      try {
+        val resultSet = statement.executeQuery("select 42")
+        resultSet.next()
+        assert(resultSet.getInt(1) == 42)
+      } finally {
+        statement.close()
+        connection.close()
+      }
+    }
+  }
+
   private def runSparkApplicationAndVerifyCompletion(
       appResource: String,
       mainClass: String,