fixing build and unit test issues

Author: ifilonenko

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/HadoopUGIUtil.scala

@@ -45,7 +45,7 @@ private[spark] class HadoopUGIUtil{
   def getCurrentTime: Long = System.currentTimeMillis()
 
    // Functions that should be in Core with Rebase to 2.3
-  @deprecated("Moved to core in 2.2", "2.2")
+  @deprecated("Moved to core in 2.3", "2.3")
   def getTokenRenewalInterval(
     renewedTokens: Iterable[Token[_ <: TokenIdentifier]],
     hadoopConf: Configuration): Option[Long] = {
@@ -62,15 +62,15 @@ private[spark] class HadoopUGIUtil{
       renewIntervals.reduceLeftOption(_ min _)
   }
 
-  @deprecated("Moved to core in 2.2", "2.2")
+  @deprecated("Moved to core in 2.3", "2.3")
   def serialize(creds: Credentials): Array[Byte] = {
     val byteStream = new ByteArrayOutputStream
     val dataStream = new DataOutputStream(byteStream)
     creds.writeTokenStorageToStream(dataStream)
     byteStream.toByteArray
   }
 
-  @deprecated("Moved to core in 2.2", "2.2")
+  @deprecated("Moved to core in 2.3", "2.3")
   def deserialize(tokenBytes: Array[Byte]): Credentials = {
     val creds = new Credentials()
     creds.readTokenStorageStream(new DataInputStream(new ByteArrayInputStream(tokenBytes)))

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KerberosTokenConfBootstrap.scala

@@ -73,4 +73,4 @@ private[spark] class KerberosTokenConfBootstrapImpl(
       pod = dtMountedPod,
       mainContainer = mainContainerWithMountedKerberos)
   }
-}
+}

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/submitsteps/hadoopsteps/HadoopStepsOrchestrator.scala

@@ -101,4 +101,4 @@ private[spark] class HadoopStepsOrchestrator(
        Seq.empty[File]
      }
   }
-}
+}

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactory.scala

@@ -47,7 +47,6 @@ private[spark] class ExecutorPodFactoryImpl(
     executorInitContainerBootstrap: Option[SparkPodInitContainerBootstrap],
     executorMountInitContainerSecretPlugin: Option[InitContainerResourceStagingServerSecretPlugin],
     executorLocalDirVolumeProvider: ExecutorLocalDirVolumeProvider,
-    shuffleManager: Option[KubernetesExternalShuffleManager],
     hadoopBootStrap: Option[HadoopConfBootstrap],
     kerberosBootstrap: Option[KerberosTokenConfBootstrap])
   extends ExecutorPodFactory {

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala

@@ -164,7 +164,6 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
         executorInitContainerBootstrap,
         executorInitContainerSecretVolumePlugin,
         executorLocalDirVolumeProvider,
-        kubernetesShuffleManager,
         hadoopBootStrap,
         kerberosBootstrap)
     val allocatorExecutor = ThreadUtils

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/DriverConfigurationStepsOrchestratorSuite.scala

@@ -18,7 +18,7 @@ package org.apache.spark.deploy.k8s.submit
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
 import org.apache.spark.deploy.k8s.config._
-import org.apache.spark.deploy.k8s.submit.submitsteps.{BaseDriverConfigurationStep, DependencyResolutionStep, DriverConfigurationStep, DriverKubernetesCredentialsStep, DriverServiceBootstrapStep, InitContainerBootstrapStep, LocalDirectoryMountConfigurationStep, MountSecretsStep, MountSmallLocalFilesStep, PythonStep, RStep}
+import org.apache.spark.deploy.k8s.submit.submitsteps.{BaseDriverConfigurationStep, DependencyResolutionStep, DriverConfigurationStep, DriverKubernetesCredentialsStep, DriverServiceBootstrapStep, HadoopConfigBootstrapStep, InitContainerBootstrapStep, LocalDirectoryMountConfigurationStep, MountSecretsStep, MountSmallLocalFilesStep, PythonStep, RStep}
 
 private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunSuite {
 
@@ -45,7 +45,8 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
         APP_NAME,
         MAIN_CLASS,
         APP_ARGS,
-        Seq.empty[String],
+        ADDITIONAL_PYTHON_FILES,
+        None,
         sparkConf)
     validateStepTypes(
         orchestrator,
@@ -69,7 +70,8 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
         APP_NAME,
         MAIN_CLASS,
         APP_ARGS,
-        Seq.empty[String],
+        ADDITIONAL_PYTHON_FILES,
+        None,
         sparkConf)
     validateStepTypes(
         orchestrator,
@@ -93,6 +95,7 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
       MAIN_CLASS,
       APP_ARGS,
       ADDITIONAL_PYTHON_FILES,
+      None,
       sparkConf)
     validateStepTypes(
         orchestrator,
@@ -116,6 +119,7 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
       MAIN_CLASS,
       APP_ARGS,
       Seq.empty[String],
+      None,
       sparkConf)
     validateStepTypes(
       orchestrator,
@@ -127,7 +131,6 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
       classOf[RStep])
   }
 
-
   test("Only local files without a resource staging server.") {
     val sparkConf = new SparkConf(false).set("spark.files", "/var/spark/file1.txt")
     val mainAppResource = JavaMainAppResource("local:///var/apps/jars/main.jar")
@@ -139,7 +142,8 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
         APP_NAME,
         MAIN_CLASS,
         APP_ARGS,
-        Seq.empty[String],
+        ADDITIONAL_PYTHON_FILES,
+        None,
         sparkConf)
     validateStepTypes(
         orchestrator,
@@ -164,7 +168,8 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
       APP_NAME,
       MAIN_CLASS,
       APP_ARGS,
-      Seq.empty[String],
+      ADDITIONAL_PYTHON_FILES,
+      None,
       sparkConf)
     validateStepTypes(
       orchestrator,
@@ -175,6 +180,31 @@ private[spark] class DriverConfigurationStepsOrchestratorSuite extends SparkFunS
       classOf[LocalDirectoryMountConfigurationStep],
       classOf[MountSecretsStep])
   }
+  test("Submission steps with hdfs interaction and HADOOP_CONF_DIR defined") {
+    val sparkConf = new SparkConf(false)
+    val mainAppResource = JavaMainAppResource("local:///var/apps/jars/main.jar")
+    val hadoopConf = Some("/etc/hadoop/conf")
+    val orchestrator = new DriverConfigurationStepsOrchestrator(
+      NAMESPACE,
+      APP_ID,
+      LAUNCH_TIME,
+      mainAppResource,
+      APP_NAME,
+      MAIN_CLASS,
+      APP_ARGS,
+      ADDITIONAL_PYTHON_FILES,
+      hadoopConf,
+      sparkConf)
+    val steps = orchestrator.getAllConfigurationSteps()
+    validateStepTypes(
+      orchestrator,
+      classOf[BaseDriverConfigurationStep],
+      classOf[DriverServiceBootstrapStep],
+      classOf[DriverKubernetesCredentialsStep],
+      classOf[DependencyResolutionStep],
+      classOf[LocalDirectoryMountConfigurationStep],
+      classOf[HadoopConfigBootstrapStep])
+  }
 
   private def validateStepTypes(
       orchestrator: DriverConfigurationStepsOrchestrator,

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodFactorySuite.scala

@@ -77,7 +77,9 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef
       None,
       None,
       None,
-      executorLocalDirVolumeProvider)
+      executorLocalDirVolumeProvider,
+      None,
+      None)
     val executor = factory.createExecutorPod(
       "1", "dummy", "dummy", Seq[(String, String)](), driverPod, Map[String, Int]())
 
@@ -117,7 +119,9 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef
         None,
         None,
         None,
-        executorLocalDirVolumeProvider)
+        executorLocalDirVolumeProvider,
+        None,
+        None)
     val executor = factory.createExecutorPod(
       "1", "dummy", "dummy", Seq[(String, String)](), driverPod, Map[String, Int]())
 
@@ -138,7 +142,9 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef
       None,
       None,
       None,
-      executorLocalDirVolumeProvider)
+      executorLocalDirVolumeProvider,
+      None,
+      None)
     val executor = factory.createExecutorPod(
       "1", "dummy", "dummy", Seq[(String, String)](), driverPod, Map[String, Int]())
 
@@ -172,7 +178,9 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef
       None,
       Some(initContainerBootstrap),
       None,
-      executorLocalDirVolumeProvider)
+      executorLocalDirVolumeProvider,
+      None,
+      None)
     val executor = factory.createExecutorPod(
       "1", "dummy", "dummy", Seq[(String, String)](), driverPod, Map[String, Int]())
 
@@ -202,7 +210,9 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef
       None,
       None,
       None,
-      executorLocalDirVolumeProvider)
+      executorLocalDirVolumeProvider,
+      None,
+      None)
     val executor = factory.createExecutorPod(
       "1", "dummy", "dummy", Seq[(String, String)](), driverPod, Map[String, Int]())
     assert(executor.getSpec.getVolumes.size === 1)
@@ -223,7 +233,9 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef
       Some(smallFiles),
       None,
       None,
-      executorLocalDirVolumeProvider)
+      executorLocalDirVolumeProvider,
+      None,
+      None)
     val executor = factory.createExecutorPod(
       "1", "dummy", "dummy", Seq[(String, String)](), driverPod, Map[String, Int]())
 
@@ -256,7 +268,9 @@ class ExecutorPodFactorySuite extends SparkFunSuite with BeforeAndAfter with Bef
         None,
         None,
         None,
-        executorLocalDirVolumeProvider)
+        executorLocalDirVolumeProvider,
+        None,
+        None)
     val executor = factory.createExecutorPod(
       "1", "dummy", "dummy", Seq[(String, String)]("qux" -> "quux"), driverPod, Map[String, Int]())