handle most PR comments

Author: ifilonenko

docs/running-on-kubernetes.md

@@ -780,7 +780,7 @@ from the other deployment modes. See the [configuration page](configuration.html
   </td>
 </tr>
 <tr>
-  <td><code>spark.kubernetes.kerberos.rewewer.principal</code></td> 
+  <td><code>spark.kubernetes.kerberos.renewer.principal</code></td> 
   <td>(none)</td>
   <td>
     Assuming you have set <code>spark.kubernetes.kerberos.enabled</code> to be true. This will let you specify 

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

@@ -47,12 +47,13 @@ private[spark] class HadoopConfBootstrapImpl(
   override def bootstrapMainContainerAndVolumes(
     originalPodWithMainContainer: PodWithMainContainer)
     : PodWithMainContainer = {
-    logInfo("HADOOP_CONF_DIR defined. Mounting HDFS specific .xml files")
-    val keyPaths = hadoopConfigFiles.map(file =>
+    logInfo("HADOOP_CONF_DIR defined. Mounting Hadoop specific .xml files")
+    val keyPaths = hadoopConfigFiles.map{ file =>
+      val fileStringPath = file.toPath.getFileName.toString
       new KeyToPathBuilder()
-        .withKey(file.toPath.getFileName.toString)
-        .withPath(file.toPath.getFileName.toString)
-      .build())
+        .withKey(fileStringPath)
+        .withPath(fileStringPath)
+      .build() }
     val hadoopSupportedPod = new PodBuilder(originalPodWithMainContainer.pod)
       .editSpec()
         .addNewVolume()
@@ -64,7 +65,7 @@ private[spark] class HadoopConfBootstrapImpl(
           .endVolume()
         .endSpec()
       .build()
-    val mainContainerWithMountedHadoopConf = new ContainerBuilder(
+    val hadoopSupportedContainer = new ContainerBuilder(
       originalPodWithMainContainer.mainContainer)
       .addNewVolumeMount()
         .withName(HADOOP_FILE_VOLUME)
@@ -81,6 +82,6 @@ private[spark] class HadoopConfBootstrapImpl(
       .build()
     originalPodWithMainContainer.copy(
       pod = hadoopSupportedPod,
-      mainContainer = mainContainerWithMountedHadoopConf)
+      mainContainer = hadoopSupportedContainer)
   }
 }

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

@@ -39,12 +39,11 @@ private[spark] class KerberosTokenConfBootstrapImpl(
   secretItemKey: String,
   userName: String) extends KerberosTokenConfBootstrap with Logging{
 
-
   override def bootstrapMainContainerAndVolumes(
   originalPodWithMainContainer: PodWithMainContainer)
   : PodWithMainContainer = {
     logInfo("Mounting HDFS DT from Secret for Secure HDFS")
-    val dtMountedPod = new PodBuilder(originalPodWithMainContainer.pod)
+    val secretMountedPod = new PodBuilder(originalPodWithMainContainer.pod)
       .editOrNewSpec()
         .addNewVolume()
           .withName(SPARK_APP_HADOOP_SECRET_VOLUME_NAME)
@@ -54,7 +53,8 @@ private[spark] class KerberosTokenConfBootstrapImpl(
           .endVolume()
         .endSpec()
       .build()
-    val mainContainerWithMountedKerberos = new ContainerBuilder(
+    // TODO: ENV_HADOOP_TOKEN_FILE_LOCATION should point to the latest token data item key.
+    val secretMountedContainer = new ContainerBuilder(
       originalPodWithMainContainer.mainContainer)
       .addNewVolumeMount()
         .withName(SPARK_APP_HADOOP_SECRET_VOLUME_NAME)
@@ -70,7 +70,7 @@ private[spark] class KerberosTokenConfBootstrapImpl(
         .endEnv()
       .build()
     originalPodWithMainContainer.copy(
-      pod = dtMountedPod,
-      mainContainer = mainContainerWithMountedKerberos)
+      pod = secretMountedPod,
+      mainContainer = secretMountedContainer)
   }
 }

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

@@ -157,12 +157,6 @@ package object config extends Logging {
       .stringConf
       .createOptional
 
-  private[spark] val KUBERNETES_SHUFFLE_DIR =
-    ConfigBuilder("spark.kubernetes.shuffle.dir")
-      .doc("Path to the shared shuffle directories.")
-      .stringConf
-      .createOptional
-
   private[spark] val KUBERNETES_SHUFFLE_APISERVER_URI =
     ConfigBuilder("spark.kubernetes.shuffle.apiServer.url")
       .doc("URL to the Kubernetes API server that the shuffle service will monitor for Spark pods.")
@@ -523,7 +517,7 @@ package object config extends Logging {
       .createOptional
 
   private[spark] val KUBERNETES_KERBEROS_RENEWER_PRINCIPAL =
-    ConfigBuilder("spark.kubernetes.kerberos.rewnewer.principal")
+    ConfigBuilder("spark.kubernetes.kerberos.renewer.principal")
       .doc("Specify the principal" +
         " you wish to renew and retrieve your Kerberos values with")
       .stringConf

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

@@ -120,6 +120,11 @@ package object constants {
   private[spark] val SPARK_HADOOP_PREFIX = "spark.hadoop."
   private[spark] val HADOOP_SECURITY_AUTHENTICATION =
     SPARK_HADOOP_PREFIX + "hadoop.security.authentication"
+
+  // Kerberos Token-Refresh Server
+  private[spark] val KERBEROS_REFRESH_LABEL_KEY = "refresh-hadoop-tokens"
+  private[spark] val KERBEROS_REFRESH_LABEL_VALUE = "yes"
+
   // Bootstrapping dependencies via a secret
   private[spark] val MOUNTED_SMALL_FILES_SECRET_MOUNT_PATH = "/etc/spark-submitted-files"
 

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

@@ -120,7 +120,12 @@ private[spark] class DriverConfigurationStepsOrchestrator(
     val hadoopConfigSteps =
       hadoopConfDir.map { conf =>
         val hadoopStepsOrchestrator =
-          new HadoopStepsOrchestrator(namespace, hadoopConfigMapName, submissionSparkConf, conf)
+          new HadoopStepsOrchestrator(
+            kubernetesResourceNamePrefix,
+            namespace,
+            hadoopConfigMapName,
+            submissionSparkConf,
+            conf)
         val hadoopConfSteps = hadoopStepsOrchestrator.getHadoopSteps()
         Some(new HadoopConfigBootstrapStep(hadoopConfSteps, hadoopConfigMapName))}
       .getOrElse(Option.empty[DriverConfigurationStep])

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

@@ -18,7 +18,8 @@ package org.apache.spark.deploy.k8s.submit.submitsteps.hadoopsteps
 
 import java.io.File
 
-import org.apache.commons.io.FileUtils.readFileToString
+import com.google.common.base.Charsets
+import com.google.common.io.Files
 
 import org.apache.spark.deploy.k8s.{HadoopConfBootstrap, PodWithMainContainer}
 import org.apache.spark.deploy.k8s.constants._
@@ -48,7 +49,7 @@ private[spark] class HadoopConfMounterStep(
        driverContainer = bootstrappedPodAndMainContainer.mainContainer,
        configMapProperties =
          hadoopConfigurationFiles.map(file =>
-           (file.toPath.getFileName.toString, readFileToString(file))).toMap,
+           (file.toPath.getFileName.toString, Files.toString(file, Charsets.UTF_8))).toMap,
        additionalDriverSparkConf = hadoopConfigSpec.additionalDriverSparkConf ++
         Map(HADOOP_CONF_DIR_LOC -> hadoopConfDir)
      )

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

@@ -46,6 +46,7 @@ import org.apache.spark.internal.Logging
   * DriverSpec.
   */
 private[spark] class HadoopKerberosKeytabResolverStep(
+  kubernetesResourceNamePrefix: String,
   submissionSparkConf: SparkConf,
   maybePrincipal: Option[String],
   maybeKeytab: Option[File],
@@ -94,12 +95,13 @@ private[spark] class HadoopKerberosKeytabResolverStep(
         hadoopUGI.getTokenRenewalInterval(tokens, hadoopConf).getOrElse(Long.MaxValue)
       val currentTime: Long = hadoopUGI.getCurrentTime
       val initialTokenDataKeyName = s"$KERBEROS_SECRET_LABEL_PREFIX-$currentTime-$renewalInterval"
-      val uniqueSecretName = s"$HADOOP_KERBEROS_SECRET_NAME.$currentTime"
+      val uniqueSecretName =
+        s"$kubernetesResourceNamePrefix-$HADOOP_KERBEROS_SECRET_NAME.$currentTime"
       val secretDT =
         new SecretBuilder()
           .withNewMetadata()
             .withName(uniqueSecretName)
-            .withLabels(Map("refresh-hadoop-tokens" -> "yes").asJava)
+            .withLabels(Map(KERBEROS_REFRESH_LABEL_KEY -> KERBEROS_REFRESH_LABEL_VALUE).asJava)
             .endMetadata()
             .addToData(initialTokenDataKeyName, Base64.encodeBase64String(data))
         .build()

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

@@ -27,10 +27,11 @@ import org.apache.spark.internal.Logging
   * Returns the complete ordered list of steps required to configure the hadoop configurations.
   */
 private[spark] class HadoopStepsOrchestrator(
+  kubernetesResourceNamePrefix: String,
   namespace: String,
   hadoopConfigMapName: String,
   submissionSparkConf: SparkConf,
-  hadoopConfDir: String) extends Logging{
+  hadoopConfDir: String) extends Logging {
    private val isKerberosEnabled = submissionSparkConf.get(KUBERNETES_KERBEROS_SUPPORT)
    private val maybePrincipal = submissionSparkConf.get(KUBERNETES_KERBEROS_PRINCIPAL)
    private val maybeKeytab = submissionSparkConf.get(KUBERNETES_KERBEROS_KEYTAB)
@@ -81,6 +82,7 @@ private[spark] class HadoopStepsOrchestrator(
           existingSecretName,
           maybeExistingSecretItemKey.get))).getOrElse(Some(
             new HadoopKerberosKeytabResolverStep(
+              kubernetesResourceNamePrefix,
               submissionSparkConf,
               maybePrincipal,
               maybeKeytab,