diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
index b524603f691a8..c030992c6d311 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -653,7 +653,9 @@ object SparkSubmit extends CommandLineUtils {
       if (args.isPython) {
         childArgs ++= Array("--primary-py-file", args.primaryResource)
         childArgs ++= Array("--main-class", "org.apache.spark.deploy.PythonRunner")
-        childArgs ++= Array("--other-py-files", args.pyFiles)
+        if (args.pyFiles != null) {
+          childArgs ++= Array("--other-py-files", args.pyFiles)
+        }
       } else {
         childArgs ++= Array("--primary-java-resource", args.primaryResource)
         childArgs ++= Array("--main-class", args.mainClass)
diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index 191ec8454b7b2..68a9334b29fc2 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -770,6 +770,22 @@ from the other deployment modes. See the [configuration page](configuration.html
     <code>myIdentifier</code>. Multiple node selector keys can be added by setting multiple configurations with this prefix.
   </td>
 </tr>
+<tr>
+  <td><code>spark.executorEnv.[EnvironmentVariableName]</code></td> 
+  <td>(none)</td>
+  <td>
+    Add the environment variable specified by <code>EnvironmentVariableName</code> to
+    the Executor process. The user can specify multiple of these to set multiple environment variables.
+  </td>
+</tr>
+<tr>
+  <td><code>spark.kubernetes.driverEnv.[EnvironmentVariableName]</code></td> 
+  <td>(none)</td>
+  <td>
+    Add the environment variable specified by <code>EnvironmentVariableName</code> to
+    the Driver process. The user can specify multiple of these to set multiple environment variables.
+  </td>
+</tr>
 </table>
 
 
diff --git a/pom.xml b/pom.xml
index b6392664e9cdc..643e40ed23f05 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2358,7 +2358,7 @@
             <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
             <junitxml>.</junitxml>
             <filereports>SparkTestSuite.txt</filereports>
-            <argLine>-ea -Xmx3g -XX:MaxPermSize=${MaxPermGen} -XX:ReservedCodeCacheSize=${CodeCacheSize} ${extraScalaTestArgs}</argLine>
+            <argLine>-ea -Xmx3g -XX:ReservedCodeCacheSize=${CodeCacheSize} ${extraScalaTestArgs}</argLine>
             <stderr/>
             <environmentVariables>
               <!--
diff --git a/resource-managers/kubernetes/README.md b/resource-managers/kubernetes/README.md
index fd1ad29eb795d..685ff343fa3be 100644
--- a/resource-managers/kubernetes/README.md
+++ b/resource-managers/kubernetes/README.md
@@ -64,10 +64,12 @@ build/mvn integration-test \
 # Running against an arbitrary cluster
 
 In order to run against any cluster, use the following:
+```sh
 build/mvn integration-test \
     -Pkubernetes -Pkubernetes-integration-tests \
     -pl resource-managers/kubernetes/integration-tests -am
     -DextraScalaTestArgs="-Dspark.kubernetes.test.master=k8s://https://<master> -Dspark.docker.test.driverImage=<driver-image> -Dspark.docker.test.executorImage=<executor-image>"
+```
 
 # Preserve the Minikube VM
 
diff --git a/resource-managers/kubernetes/architecture-docs/external-shuffle-service.md b/resource-managers/kubernetes/architecture-docs/external-shuffle-service.md
new file mode 100644
index 0000000000000..e303800943d42
--- /dev/null
+++ b/resource-managers/kubernetes/architecture-docs/external-shuffle-service.md
@@ -0,0 +1,29 @@
+---
+layout: global
+title: Kubernetes Implementation of the External Shuffle Service
+---
+
+# External Shuffle Service
+
+The `KubernetesExternalShuffleService` was added to allow Spark to use Dynamic Allocation Mode when
+running in Kubernetes. The shuffle service is responsible for persisting shuffle files beyond the
+lifetime of the executors, allowing the number of executors to scale up and down without losing
+computation. 
+
+The implementation of choice is as a DaemonSet that runs a shuffle-service pod on each node.
+Shuffle-service pods and executors pods that land on the same node share disk using hostpath
+volumes. Spark requires that each executor must know the IP address of the shuffle-service pod that
+shares disk with it. 
+
+The user specifies the shuffle service pods they want executors of a particular SparkJob to use
+through two new properties:
+
+* spark.kubernetes.shuffle.service.labels
+* spark.kubernetes.shuffle.namespace
+
+KubernetesClusterSchedulerBackend is aware of shuffle service pods and the node corresponding to
+them in a particular namespace. It uses this data to configure the executor pods to connect with the
+shuffle services that are co-located with them on the same node. 
+
+There is additional logic in the `KubernetesExternalShuffleService` to watch the Kubernetes API,
+detect failures, and proactively cleanup files in those error cases. 
diff --git a/resource-managers/kubernetes/architecture-docs/scheduler-backend.md b/resource-managers/kubernetes/architecture-docs/scheduler-backend.md
new file mode 100644
index 0000000000000..cc43bd7816a8b
--- /dev/null
+++ b/resource-managers/kubernetes/architecture-docs/scheduler-backend.md
@@ -0,0 +1,49 @@
+---
+layout: global
+title: Kubernetes Implementation of the Spark Scheduler Backend
+---
+
+# Scheduler Backend
+
+The general idea is to run Spark drivers and executors inside Kubernetes [Pods](https://kubernetes.io/docs/concepts/workloads/pods/pod/). 
+Pods are a co-located and co-scheduled group of one or more containers run in a shared context. The main component is KubernetesClusterSchedulerBackend, 
+an implementation of CoarseGrainedSchedulerBackend, which manages allocating and destroying executors via the Kubernetes API. 
+There are auxiliary and optional components: `ResourceStagingServer` and `KubernetesExternalShuffleService`, which serve specific purposes described further below.
+
+The scheduler backend is invoked in the driver associated with a particular job. The driver may run outside the cluster (client mode) or within (cluster mode). 
+The scheduler backend manages [pods](http://kubernetes.io/docs/user-guide/pods/) for each executor. 
+The executor code is running within a Kubernetes pod, but remains unmodified and unaware of the orchestration layer. 
+When a job is running, the scheduler backend configures and creates executor pods with the following properties:
+
+- The pod's container runs a pre-built Docker image containing a Spark distribution (with Kubernetes integration) and 
+invokes the Java runtime with the CoarseGrainedExecutorBackend main class.
+- The scheduler backend specifies environment variables on the executor pod to configure its runtime, p
+articularly for its JVM options, number of cores, heap size, and the driver's hostname.
+- The executor container has [resource limits and requests](https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/#resource-requests-and-limits-of-pod-and-container) 
+that are set in accordance to the resource limits specified in the Spark configuration (executor.cores and executor.memory in the application's SparkConf)
+- The executor pods may also be launched into a particular [Kubernetes namespace](https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/%5C), 
+or target a particular subset of nodes in the Kubernetes cluster, based on the Spark configuration supplied.
+
+## Requesting Executors
+
+Spark requests for new executors through the `doRequestTotalExecutors(numExecutors: Int)` method. 
+The scheduler backend keeps track of the request made by Spark core for the number of executors.
+
+A separate kubernetes-pod-allocator thread handles the creation of new executor pods with appropriate throttling and monitoring. 
+This indirection is required because the Kubernetes API Server accepts requests for new executor pods optimistically, with the 
+anticipation of being able to eventually run them. However, it is undesirable to have a very large number of pods that cannot be 
+scheduled and stay pending within the cluster. Hence, the kubernetes-pod-allocator uses the Kubernetes API to make a decision to 
+submit new requests for executors based on whether previous pod creation requests have completed. This gives us control over how 
+fast a job scales up (which can be configured), and helps prevent Spark jobs from DOS-ing the Kubernetes API server with pod creation requests.
+
+## Destroying Executors
+
+Spark requests deletion of executors through the `doKillExecutors(executorIds: List[String])`
+method.
+
+The inverse behavior is required in the implementation of doKillExecutors(). When the executor
+allocation manager desires to remove executors from the application, the scheduler should find the
+pods that are running the appropriate executors, and tell the API server to stop these pods.
+It's worth noting that this code does not have to decide on the executors that should be
+removed. When `doKillExecutors()` is called, the executors that are to be removed have already been
+selected by the CoarseGrainedSchedulerBackend and ExecutorAllocationManager.
diff --git a/resource-managers/kubernetes/architecture-docs/submission-client.md b/resource-managers/kubernetes/architecture-docs/submission-client.md
new file mode 100644
index 0000000000000..022cf15c60be7
--- /dev/null
+++ b/resource-managers/kubernetes/architecture-docs/submission-client.md
@@ -0,0 +1,231 @@
+---
+layout: global
+title: Implementation of Submitting Applications to Kubernetes
+---
+
+
+Similarly to YARN and Standalone mode, it is common for Spark applications to be deployed on Kubernetes through the
+`spark-submit` process. Applications are deployed on Kubernetes via sending YAML files to the Kubernetes API server.
+These YAML files declare the structure and behavior of the processes that will be run. However, such a declarative
+approach to application deployment differs considerably from how Spark applications are deployed via the `spark-submit`
+API. There are contracts provided by `spark-submit` that should work in Kubernetes in a consistent manner to the other
+cluster managers that `spark-submit` can deploy on.
+
+This document outlines the design of the **Kubernetes submission client**, which effectively serves as a *translation
+of options provided in spark-submit to a specification of one or more Kubernetes API resources that represent the
+Spark driver*.
+
+# Entry Point
+
+As with the other cluster managers, the user's invocation of `spark-submit` will eventually delegate to running
+`org.apache.spark.deploy.SparkSubmit#submit`. This method calls a main method that handles the submission logic
+for a specific type of cluster manager. The top level entry point for the Kubernetes submission logic is in
+`org.apache.spark.deploy.kubernetes.submit.Client#main()`.
+
+# Driver Configuration Steps
+
+In order to render submission parameters into the final Kubernetes driver pod specification, and do it in a scalable
+manner, the submission client breaks pod construction down into a
+series of configuration steps, each of which is responsible for handling some specific aspect of configuring the driver.
+A top level component then iterates through all of the steps to produce a final set of Kubernetes resources that are
+then deployed on the cluster.
+
+## Interface Definitions
+
+More formally, a configuration step must implement the following trait:
+
+```scala
+package org.apache.spark.deploy.kubernetes.submit.submitsteps
+
+/**
+ * Represents a step in preparing the Kubernetes driver.
+ */
+private[spark] trait DriverConfigurationStep {
+
+  /**
+   * Apply some transformation to the previous state of the driver to add a new feature to it.
+   */
+  def configureDriver(driverSpec: KubernetesDriverSpec): KubernetesDriverSpec
+}
+```
+
+A `DriverConfigurationStep` is thus a function that transforms a `KubernetesDriverSpec` into another
+`KubernetesDriverSpec`, by taking the original specification and making additions to the specification in accordance to
+the specific feature that step is responsible for. A `KubernetesDriverSpec` is a data structure with the following
+properties:
+
+```scala
+private[spark] case class KubernetesDriverSpec(
+    driverPod: Pod,
+    driverContainer: Container,
+    otherKubernetesResources: Seq[HasMetadata],
+    driverSparkConf: SparkConf)
+```
+
+The `Pod` and `Container` classes are Java representations of Kubernetes pods and containers respectively, and the
+`HasMetadata` type corresponds to an arbitrary Kubernetes API resource such as a `Secret` or a `ConfigMap`. Kubernetes
+primitives are represented using an [open-source Java Kubernetes client](https://github.com/fabric8io/kubernetes-client).
+The `otherKubernetesResources` field represents Kubernetes resources that are required by the Spark application. For
+example, the driver may require a `ConfigMap` or `Secret` resource to be created that will be mounted into the driver
+container.
+
+## Requirements for Configuration Steps
+
+Configuration steps must be *independent*. A given configuration step should not be opinionated about the other
+configuration steps that are executed before or after it. By extension, configuration steps should be *strictly
+additive*. A given configuration step should not attempt to mutate an existing field nor remove fields set in the
+input driver specification.
+
+## Composition of Configuration Steps
+
+Finally, configuration steps are wired together by an **orchestrator**. The orchestrator effectively translates the
+parameters sent to `spark-submit` into the set of steps required to configure the final `KubernetesDriverSpec`. The
+top level submission client takes the final `KubernetesDriverSpec` object and builds the final requests to the
+Kubernetes API server to deploy the Kubernetes resources that comprise the Spark driver. The top level submission
+process can thus be expressed as follows in pseudo-code with roughly Scala syntax:
+
+```scala
+def runApplication(sparkSubmitArguments: SparkSubmitArguments) {
+  val initialSpec = createEmptyDriverSpec()
+  val orchestrator = new DriverConfigurationStepsOrchestrator(sparkSubmitArguments)
+  val steps = orchestrator.getSubmissionSteps()
+  var currentSpec = initialSpec
+  // iteratively apply the configuration steps to build up the pod spec:
+  for (step <- steps) {
+    currentSpec = step.configureDriver(currentSpec)
+  }
+  // Put the container in the pod spec
+  val resolvedPod = attachContainer(currentSpec.driverPod, currentSpec.driverContainer)
+  kubernetes.create(resolvedPod + currentSpec.otherKubernetesResources)
+}
+```
+
+## Writing a New Configuration Step
+
+All configuration steps should be placed in the `org.apache.spark.deploy.kubernetes.submit.submitsteps` package.
+Examples of other configuration steps can be found in this package as well. Ensure that the new configuration step is
+returned in `org.apache.spark.deploy.kubernetes.submit.DriverConfigurationStepsOrchestrator#getAllConfigurationSteps()`.
+
+# Dependency Management
+
+Spark applications typically depend on binaries and various configuration files which are hosted in various locations.
+Kubernetes applications typically bundle binary dependencies such as jars inside Docker images. However, Spark's API
+fundamentally allows dependencies to be provided from many other locations, including the submitter's local disk.
+These dependencies have to be deployed into the driver and executor containers before they run. This is challenging
+because unlike Hadoop YARN which requires co-deployment with an HDFS cluster, Kubernetes clusters do not have a
+large-scale persistent storage layer that would be available across every Kubernetes cluster.
+
+## Resource Staging Server
+
+The *resource staging server* is a lightweight daemon that serves as a file store for application dependencies. It has
+two endpoints which effectively correspond to putting files into the server and getting files out of the server. When
+files are put into the server, the server returns a unique identifier and a secret token in the response to the client.
+This identifier and secret token must be provided when a client makes a request to retrieve the files that were uploaded
+to the server.
+
+### Resource Staging Server API Definition
+
+The resource staging server has the following Scala API which would then be translated into HTTP endpoints via Jetty and
+JAX-RS. Associated structures passed as input and output are also defined below:
+
+```scala
+private[spark] trait ResourceStagingService {
+
+  def uploadResources(resources: InputStream, resourcesOwner: StagedResourcesOwner): SubmittedResourceIdAndSecret
+  def downloadResources(resourceId: String, resourceSecret: String): StreamingOutput
+}
+
+case class StagedResourcesOwner(
+    ownerNamespace: String,
+    ownerLabels: Map[String, String],
+    ownerType: StagedResourcesOwnerType.OwnerType)
+    
+// Pseudo-code to represent an enum
+enum StagedResourcesOwnerType.OwnerType = { Pod }
+
+case class SubmittedResourceIdAndSecret(resourceId: String, resourceSecret: String)
+```
+
+Clients that send resources to the server do so in a streaming manner so that both the server and the client do not
+need to hold the entire resource bundle in memory. Aside from the notion of the `StagedResourcesOwner` that is provided
+on uploads and not for downloads, uploading is symmetrical to downloading. The significance of the
+`StagedResourcesOwner` is discussed below.
+
+### Cleaning Up Stale Resources
+
+The resource staging server is built to provide resources for the pods and containers in a Kubernetes cluster. These
+pods are ephemeral, so at some point there will be no need for the resources that were sent for a specific application.
+Clients indicate the set of resources that would be using a given resource bundle by providing a description of the
+resource's "owner". The `StagedResourceOwner` is this description, defining the owner as a Kubernetes API object in
+a given namespace and having a specific set of labels.
+
+The resource staging server keeps track of the resources that were sent to it. When the resource is first uploaded, it
+is marked as "unused". If the resource remains unused for a period of time, it is cleaned up. A resource is marked as
+"used" when a request is made to download it. After that, the server periodically checks the API server to see if any
+Kubernetes API objects exist that match the description of the owner. If no such objects exist, then resource staging
+server cleans up the uploaded resource. See `org.apache.spark.deploy.rest.kubernetes.StagedResourcesCleaner` for the
+code that manages the resource's lifecycle.
+
+A resource owner can currently only be a pod, but hypothetically one could want to tie the lifetime of a resource to the
+lifetime of many pods under a higher level Kubernetes object like a Deployment or a StatefulSet, all of which depend on
+the uploaded resource. The resource staging server's API can be extended to tie ownership of a resource to any
+Kubernetes API object type, as long as we update the `StagedResourcesOwnerType` enumeration accordingly.
+
+### Usage in Spark
+
+Spark-submit supports adding jars and files by passing `--jars` and `--files` to `spark-submit` respectively. The spark
+configurations `spark.jars` and `spark.files` can also be set to provide this information. The submission client
+determines the list of jars and files that the application needs, and it determines if any of them are files being sent
+from the submitter's local machine. If any files are being sent from the local machine, the user must have specified a
+URL for the resource staging server to send the files to.
+
+Local jars and files are compacted into a tarball which are then uploaded to the resource staging server. The submission
+client then knows the secret token that the driver and executors must use to download the files again. These secrets
+are mounted into an [init-container](https://kubernetes.io/docs/concepts/workloads/pods/init-containers/)
+that runs before the driver and executor processes run, and the init-container
+downloads the uploaded resources from the resource staging server.
+
+### Other Considered Alternatives
+
+The resource staging server was considered the best option among other alternative solutions to this problem.
+
+A first implementation effectively included the resource staging server in the driver container itself. The driver
+container ran a custom command that opened an HTTP endpoint and waited for the submission client to send resources to
+it. The server would then run the driver application after it had received the resources from the user's local
+machine. The problem with this approach is that the submission client needs to deploy the driver in such a way that the
+driver itself would be reachable from outside of the cluster, but it is difficult for an automated framework which is
+not aware of the cluster's configuration to expose an arbitrary pod in a generic way. The resource staging server allows
+a cluster administrator to expose the resource staging server in a manner that makes sense for their cluster, such as
+with an Ingress or with a NodePort service.
+
+It is also impossible to use Kubernetes API objects like Secrets or ConfigMaps to store application binaries. These
+objects require their contents to be small so that they can fit in etcd.
+
+Finally, as mentioned before, the submission client should not be opinionated about storing dependencies in a
+distributed storage system like HDFS, because not all Kubernetes clusters will have the same types of persistent storage
+layers. Spark supports fetching jars directly from distributed storage layers though, so users can feel free to manually
+push their dependencies to their appropriate systems and refer to them by their remote URIs in the submission request.
+
+## Init-Containers
+
+The driver pod and executor pods both use [init-containers](https://kubernetes.io/docs/concepts/workloads/pods/init-containers/)
+to localize resources before the driver and
+executor processes launch. As mentioned before, the init-container fetches dependencies from the resource staging
+server. However, even if the resource staging server is not being used, files still need to be localized from remote
+locations such as HDFS clusters or HTTP file servers. The init-container will fetch these dependencies accordingly as
+well.
+
+Init-containers were preferred over fetching the dependencies in the main container primarily because this allows the
+main container's runtime commands to be simplified. Using init-containers to fetch these remote dependencies allows the
+main image command to simply be an invocation of Java that runs the user's main class directly. The execution of the
+file localizer process can also be shared by both the driver and the executor images without needing to be copied
+into both image commands. Finally, it becomes easier to debug localization failures as they will be easily spotted as
+being a failure in the pod's initialization lifecycle phase.
+
+# Future Work
+
+* The driver's pod specification should be highly customizable, to the point where users may want to specify a template
+pod spec in a YAML file: https://github.com/apache-spark-on-k8s/spark/issues/38.
+* The resource staging server can be backed by a distributed file store like HDFS to improve robustness and scalability.
+* Additional driver bootstrap steps need to be added to support communication with Kerberized HDFS clusters:
+  * https://github.com/apache-spark-on-k8s/spark/pull/391
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/config.scala
index c6772c1cb5ae4..6e1633f6a63cb 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/config.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/config.scala
@@ -126,6 +126,8 @@ package object config extends Logging {
       .stringConf
       .createOptional
 
+  private[spark] val KUBERNETES_DRIVER_ENV_KEY = "spark.kubernetes.driverEnv."
+
   private[spark] val KUBERNETES_DRIVER_ANNOTATIONS =
     ConfigBuilder("spark.kubernetes.driver.annotations")
       .doc("Custom annotations that will be added to the driver pod. This should be a" +
@@ -491,6 +493,19 @@ package object config extends Logging {
       .stringConf
       .createOptional
 
+  private[spark] val KUBERNETES_DRIVER_CLUSTER_NODENAME_DNS_LOOKUP_ENABLED =
+    ConfigBuilder("spark.kubernetes.driver.hdfslocality.clusterNodeNameDNSLookup.enabled")
+      .doc("Whether or not HDFS locality support code should look up DNS for full hostnames of" +
+        " cluster nodes. In some K8s clusters, notably GKE, cluster node names are short" +
+        " hostnames, and so comparing them against HDFS datanode hostnames always fail. To fix," +
+        " enable this flag. This is disabled by default because DNS lookup can be expensive." +
+        " The driver can slow down and fail to respond to executor heartbeats in time." +
+        " If enabling this flag, make sure your DNS server has enough capacity" +
+        " for the workload.")
+      .internal()
+      .booleanConf
+      .createWithDefault(false)
+
   private[spark] val KUBERNETES_EXECUTOR_LIMIT_CORES =
     ConfigBuilder("spark.kubernetes.executor.limit.cores")
       .doc("Specify the hard cpu limit for a single executor pod")
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala
index 92f051b2ac298..a377cc4942ad4 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala
@@ -69,6 +69,7 @@ package object constants {
   private[spark] val ENV_MOUNTED_FILES_DIR = "SPARK_MOUNTED_FILES_DIR"
   private[spark] val ENV_PYSPARK_FILES = "PYSPARK_FILES"
   private[spark] val ENV_PYSPARK_PRIMARY = "PYSPARK_PRIMARY"
+  private[spark] val ENV_JAVA_OPT_PREFIX = "SPARK_JAVA_OPT_"
 
   // Bootstrapping dependencies with the init-container
   private[spark] val INIT_CONTAINER_ANNOTATION = "pod.beta.kubernetes.io/init-containers"
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/Client.scala
index 2fa9b416330e5..1901f1198a84f 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/Client.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/Client.scala
@@ -18,9 +18,10 @@ package org.apache.spark.deploy.kubernetes.submit
 
 import java.util.{Collections, UUID}
 
-import io.fabric8.kubernetes.api.model.{ContainerBuilder, OwnerReferenceBuilder, PodBuilder}
+import io.fabric8.kubernetes.api.model.{ContainerBuilder, EnvVar, EnvVarBuilder, OwnerReferenceBuilder, PodBuilder}
 import io.fabric8.kubernetes.client.KubernetesClient
 import scala.collection.mutable
+import scala.collection.JavaConverters._
 
 import org.apache.spark.SparkConf
 import org.apache.spark.deploy.kubernetes.config._
@@ -92,18 +93,21 @@ private[spark] class Client(
       currentDriverSpec = nextStep.configureDriver(currentDriverSpec)
     }
     val resolvedDriverJavaOpts = currentDriverSpec
-      .driverSparkConf
-      // We don't need this anymore since we just set the JVM options on the environment
-      .remove(org.apache.spark.internal.config.DRIVER_JAVA_OPTIONS)
-      .getAll
-      .map {
-        case (confKey, confValue) => s"-D$confKey=$confValue"
-      }.mkString(" ") + driverJavaOptions.map(" " + _).getOrElse("")
+        .driverSparkConf
+        // We don't need this anymore since we just set the JVM options on the environment
+        .remove(org.apache.spark.internal.config.DRIVER_JAVA_OPTIONS)
+        .getAll
+        .map {
+          case (confKey, confValue) => s"-D$confKey=$confValue"
+        } ++ driverJavaOptions.map(Utils.splitCommandString).getOrElse(Seq.empty)
+    val driverJavaOptsEnvs: Seq[EnvVar] = resolvedDriverJavaOpts.zipWithIndex.map {
+      case (option, index) => new EnvVarBuilder()
+          .withName(s"$ENV_JAVA_OPT_PREFIX$index")
+          .withValue(option)
+          .build()
+    }
     val resolvedDriverContainer = new ContainerBuilder(currentDriverSpec.driverContainer)
-      .addNewEnv()
-        .withName(ENV_DRIVER_JAVA_OPTS)
-        .withValue(resolvedDriverJavaOpts)
-        .endEnv()
+      .addAllToEnv(driverJavaOptsEnvs.asJava)
       .build()
     val resolvedDriverPod = new PodBuilder(currentDriverSpec.driverPod)
       .editSpec()
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/BaseDriverConfigurationStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/BaseDriverConfigurationStep.scala
index b3f509b44054e..a8539e0772163 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/BaseDriverConfigurationStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/BaseDriverConfigurationStep.scala
@@ -72,6 +72,13 @@ private[spark] class BaseDriverConfigurationStep(
     require(!driverCustomAnnotations.contains(SPARK_APP_NAME_ANNOTATION),
         s"Annotation with key $SPARK_APP_NAME_ANNOTATION is not allowed as it is reserved for" +
             s" Spark bookkeeping operations.")
+
+    val driverCustomEnvs = submissionSparkConf.getAllWithPrefix(KUBERNETES_DRIVER_ENV_KEY).toSeq
+      .map(env => new EnvVarBuilder()
+        .withName(env._1)
+        .withValue(env._2)
+        .build())
+
     val allDriverAnnotations = driverCustomAnnotations ++ Map(SPARK_APP_NAME_ANNOTATION -> appName)
     val nodeSelector = ConfigurationUtils.parsePrefixedKeyValuePairs(
       submissionSparkConf, KUBERNETES_NODE_SELECTOR_PREFIX, "node selector")
@@ -91,6 +98,7 @@ private[spark] class BaseDriverConfigurationStep(
       .withName(DRIVER_CONTAINER_NAME)
       .withImage(driverDockerImage)
       .withImagePullPolicy(dockerImagePullPolicy)
+      .addAllToEnv(driverCustomEnvs.asJava)
       .addToEnv(driverExtraClasspathEnv.toSeq: _*)
       .addNewEnv()
         .withName(ENV_DRIVER_MEMORY)
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/InetAddressUtil.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/InetAddressUtil.scala
new file mode 100644
index 0000000000000..e04ab9e541963
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/InetAddressUtil.scala
@@ -0,0 +1,36 @@
+/*
+ * 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.scheduler.cluster.kubernetes
+
+import java.net.InetAddress
+
+/**
+ * Gets full host names of given IP addresses from DNS.
+ */
+private[kubernetes] trait InetAddressUtil {
+
+  def getFullHostName(ipAddress: String): String
+}
+
+private[kubernetes] object InetAddressUtilImpl extends InetAddressUtil {
+
+  // NOTE: This does issue a network call to DNS. Caching is done internally by the InetAddress
+  // class for both hits and misses.
+  override def getFullHostName(ipAddress: String): String = {
+    InetAddress.getByName(ipAddress).getCanonicalHostName
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala
index c246b597b6f0f..06de690800492 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala
@@ -18,18 +18,20 @@ package org.apache.spark.scheduler.cluster.kubernetes
 
 import java.io.Closeable
 import java.net.InetAddress
-import java.util.concurrent.TimeUnit
+import java.util.Collections
+import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
 import java.util.concurrent.atomic.{AtomicInteger, AtomicLong, AtomicReference}
 
+import scala.collection.{concurrent, mutable}
+import scala.collection.JavaConverters._
+import scala.concurrent.{ExecutionContext, Future}
+
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.scala.DefaultScalaModule
-import io.fabric8.kubernetes.api.model.{ContainerBuilder, ContainerPortBuilder, EnvVarBuilder, EnvVarSourceBuilder, Pod, PodBuilder, QuantityBuilder}
+import io.fabric8.kubernetes.api.model._
 import io.fabric8.kubernetes.client.{KubernetesClient, KubernetesClientException, Watcher}
 import io.fabric8.kubernetes.client.Watcher.Action
 import org.apache.commons.io.FilenameUtils
-import scala.collection.JavaConverters._
-import scala.collection.mutable
-import scala.concurrent.{ExecutionContext, Future}
 
 import org.apache.spark.{SparkContext, SparkEnv, SparkException}
 import org.apache.spark.deploy.kubernetes.{ConfigurationUtils, InitContainerResourceStagingServerSecretPlugin, PodWithDetachedInitContainer, SparkPodInitContainerBootstrap}
@@ -39,8 +41,8 @@ import org.apache.spark.deploy.kubernetes.submit.InitContainerUtil
 import org.apache.spark.internal.config
 import org.apache.spark.network.netty.SparkTransportConf
 import org.apache.spark.network.shuffle.kubernetes.KubernetesExternalShuffleClient
-import org.apache.spark.rpc.{RpcCallContext, RpcEndpointAddress, RpcEnv}
-import org.apache.spark.scheduler.TaskSchedulerImpl
+import org.apache.spark.rpc.{RpcAddress, RpcCallContext, RpcEndpointAddress, RpcEnv}
+import org.apache.spark.scheduler.{ExecutorExited, SlaveLost, TaskSchedulerImpl}
 import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{RetrieveSparkAppConfig, SparkAppConfig}
 import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
 import org.apache.spark.util.{ThreadUtils, Utils}
@@ -56,10 +58,18 @@ private[spark] class KubernetesClusterSchedulerBackend(
   import KubernetesClusterSchedulerBackend._
 
   private val RUNNING_EXECUTOR_PODS_LOCK = new Object
-  private val runningExecutorPods = new mutable.HashMap[String, Pod] // Indexed by executor IDs.
-
+  // Indexed by executor IDs and guarded by RUNNING_EXECUTOR_PODS_LOCK.
+  private val runningExecutorsToPods = new mutable.HashMap[String, Pod]
+  // Indexed by executor pod names and guarded by RUNNING_EXECUTOR_PODS_LOCK.
+  private val runningPodsToExecutors = new mutable.HashMap[String, String]
+  // TODO(varun): Get rid of this lock object by my making the underlying map a concurrent hash map.
   private val EXECUTOR_PODS_BY_IPS_LOCK = new Object
-  private val executorPodsByIPs = new mutable.HashMap[String, Pod] // Indexed by executor IP addrs.
+  // Indexed by executor IP addrs and guarded by EXECUTOR_PODS_BY_IPS_LOCK
+  private val executorPodsByIPs = new mutable.HashMap[String, Pod]
+  private val failedPods: concurrent.Map[String, ExecutorExited] = new
+      ConcurrentHashMap[String, ExecutorExited]().asScala
+  private val executorsToRemove = Collections.newSetFromMap[String](
+    new ConcurrentHashMap[String, java.lang.Boolean]()).asScala
 
   private val executorExtraClasspath = conf.get(
     org.apache.spark.internal.config.EXECUTOR_CLASS_PATH)
@@ -136,7 +146,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
       val parsedShuffleLabels = ConfigurationUtils.parseKeyValuePairs(
         conf.get(KUBERNETES_SHUFFLE_LABELS), KUBERNETES_SHUFFLE_LABELS.key,
             "shuffle-labels")
-      if (parsedShuffleLabels.size == 0) {
+      if (parsedShuffleLabels.isEmpty) {
         throw new SparkException(s"Dynamic allocation enabled " +
           s"but no ${KUBERNETES_SHUFFLE_LABELS.key} specified")
       }
@@ -171,12 +181,13 @@ private[spark] class KubernetesClusterSchedulerBackend(
   private val executorWatchResource = new AtomicReference[Closeable]
   protected var totalExpectedExecutors = new AtomicInteger(0)
 
+
   private val driverUrl = RpcEndpointAddress(
     sc.getConf.get("spark.driver.host"),
     sc.getConf.getInt("spark.driver.port", DEFAULT_DRIVER_PORT),
     CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString
 
-  private val initialExecutors = getInitialTargetExecutorNumber(1)
+  private val initialExecutors = getInitialTargetExecutorNumber()
 
   private val podAllocationInterval = conf.get(KUBERNETES_ALLOCATION_BATCH_DELAY)
   require(podAllocationInterval > 0, s"Allocation batch delay " +
@@ -193,23 +204,74 @@ private[spark] class KubernetesClusterSchedulerBackend(
 
   private val allocatorRunnable: Runnable = new Runnable {
 
+    // Number of times we are allowed check for the loss reason for an executor before we give up
+    // and assume the executor failed for good, and attribute it to a framework fault.
+    private val MAX_EXECUTOR_LOST_REASON_CHECKS = 10
+    private val executorsToRecover = new mutable.HashSet[String]
+    // Maintains a map of executor id to count of checks performed to learn the loss reason
+    // for an executor.
+    private val executorReasonChecks = new mutable.HashMap[String, Int]
+
     override def run(): Unit = {
-      if (totalRegisteredExecutors.get() < runningExecutorPods.size) {
-        logDebug("Waiting for pending executors before scaling")
-      } else if (totalExpectedExecutors.get() <= runningExecutorPods.size) {
-        logDebug("Maximum allowed executor limit reached. Not scaling up further.")
-      } else {
-        val nodeToLocalTaskCount = getNodesWithLocalTaskCounts
-        RUNNING_EXECUTOR_PODS_LOCK.synchronized {
+      removeFailedExecutors()
+      RUNNING_EXECUTOR_PODS_LOCK.synchronized {
+        if (totalRegisteredExecutors.get() < runningExecutorsToPods.size) {
+          logDebug("Waiting for pending executors before scaling")
+        } else if (totalExpectedExecutors.get() <= runningExecutorsToPods.size) {
+          logDebug("Maximum allowed executor limit reached. Not scaling up further.")
+        } else {
+          val nodeToLocalTaskCount = getNodesWithLocalTaskCounts
           for (i <- 0 until math.min(
-            totalExpectedExecutors.get - runningExecutorPods.size, podAllocationSize)) {
-            runningExecutorPods += allocateNewExecutorPod(nodeToLocalTaskCount)
+            totalExpectedExecutors.get - runningExecutorsToPods.size, podAllocationSize)) {
+            val (executorId, pod) = allocateNewExecutorPod(nodeToLocalTaskCount)
+            runningExecutorsToPods.put(executorId, pod)
+            runningPodsToExecutors.put(pod.getMetadata.getName, executorId)
             logInfo(
-              s"Requesting a new executor, total executors is now ${runningExecutorPods.size}")
+              s"Requesting a new executor, total executors is now ${runningExecutorsToPods.size}")
           }
         }
       }
     }
+
+    def removeFailedExecutors(): Unit = {
+      val localRunningExecutorsToPods = RUNNING_EXECUTOR_PODS_LOCK.synchronized {
+        runningExecutorsToPods.toMap
+      }
+      executorsToRemove.foreach { case (executorId) =>
+        localRunningExecutorsToPods.get(executorId).map { pod: Pod =>
+          failedPods.get(pod.getMetadata.getName).map { executorExited: ExecutorExited =>
+            logDebug(s"Removing executor $executorId with loss reason " + executorExited.message)
+            removeExecutor(executorId, executorExited)
+            if (!executorExited.exitCausedByApp) {
+              executorsToRecover.add(executorId)
+            }
+          }.getOrElse(removeExecutorOrIncrementLossReasonCheckCount(executorId))
+        }.getOrElse(removeExecutorOrIncrementLossReasonCheckCount(executorId))
+
+        executorsToRecover.foreach(executorId => {
+          executorsToRemove -= executorId
+          executorReasonChecks -= executorId
+          RUNNING_EXECUTOR_PODS_LOCK.synchronized {
+            runningExecutorsToPods.remove(executorId).map { pod: Pod =>
+              kubernetesClient.pods().delete(pod)
+              runningPodsToExecutors.remove(pod.getMetadata.getName)
+            }.getOrElse(logWarning(s"Unable to remove pod for unknown executor $executorId"))
+          }
+        })
+        executorsToRecover.clear()
+      }
+    }
+
+    def removeExecutorOrIncrementLossReasonCheckCount(executorId: String): Unit = {
+      val reasonCheckCount = executorReasonChecks.getOrElse(executorId, 0)
+      if (reasonCheckCount > MAX_EXECUTOR_LOST_REASON_CHECKS) {
+        removeExecutor(executorId, SlaveLost("Executor lost for unknown reasons"))
+        executorsToRecover.add(executorId)
+        executorReasonChecks -= executorId
+      } else {
+        executorReasonChecks.put(executorId, reasonCheckCount + 1)
+      }
+    }
   }
 
   private val objectMapper = new ObjectMapper().registerModule(DefaultScalaModule)
@@ -281,8 +343,9 @@ private[spark] class KubernetesClusterSchedulerBackend(
     // indication as to why.
     try {
       RUNNING_EXECUTOR_PODS_LOCK.synchronized {
-        runningExecutorPods.values.foreach(kubernetesClient.pods().delete(_))
-        runningExecutorPods.clear()
+        runningExecutorsToPods.values.foreach(kubernetesClient.pods().delete(_))
+        runningExecutorsToPods.clear()
+        runningPodsToExecutors.clear()
       }
       EXECUTOR_PODS_BY_IPS_LOCK.synchronized {
         executorPodsByIPs.clear()
@@ -394,7 +457,16 @@ private[spark] class KubernetesClusterSchedulerBackend(
         .withValue(cp)
         .build()
     }
-    val requiredEnv = Seq(
+    val executorExtraJavaOptionsEnv = conf
+        .get(org.apache.spark.internal.config.EXECUTOR_JAVA_OPTIONS)
+        .map { opts =>
+          val delimitedOpts = Utils.splitCommandString(opts)
+          delimitedOpts.zipWithIndex.map {
+            case (opt, index) =>
+              new EnvVarBuilder().withName(s"$ENV_JAVA_OPT_PREFIX$index").withValue(opt).build()
+          }
+        }.getOrElse(Seq.empty[EnvVar])
+    val executorEnv = (Seq(
       (ENV_EXECUTOR_PORT, executorPort.toString),
       (ENV_DRIVER_URL, driverUrl),
       // Executor backend expects integral value for executor cores, so round it up to an int.
@@ -402,7 +474,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
       (ENV_EXECUTOR_MEMORY, executorMemoryString),
       (ENV_APPLICATION_ID, applicationId()),
       (ENV_EXECUTOR_ID, executorId),
-      (ENV_MOUNTED_CLASSPATH, s"$executorJarsDownloadDir/*"))
+      (ENV_MOUNTED_CLASSPATH, s"$executorJarsDownloadDir/*")) ++ sc.executorEnvs.toSeq)
       .map(env => new EnvVarBuilder()
         .withName(env._1)
         .withValue(env._2)
@@ -414,7 +486,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
           .withNewFieldRef("v1", "status.podIP")
           .build())
         .build()
-      )
+      ) ++ executorExtraJavaOptionsEnv ++ executorExtraClasspathEnv.toSeq
     val requiredPorts = Seq(
       (EXECUTOR_PORT_NAME, executorPort),
       (BLOCK_MANAGER_PORT_NAME, blockmanagerPort))
@@ -434,8 +506,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
         .addToLimits("memory", executorMemoryLimitQuantity)
         .addToRequests("cpu", executorCpuQuantity)
       .endResources()
-      .addAllToEnv(requiredEnv.asJava)
-      .addToEnv(executorExtraClasspathEnv.toSeq: _*)
+      .addAllToEnv(executorEnv.asJava)
       .withPorts(requiredPorts.asJava)
       .build()
 
@@ -455,6 +526,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
       .endMetadata()
       .withNewSpec()
         .withHostname(hostname)
+        .withRestartPolicy("Never")
         .withNodeSelector(nodeSelector.asJava)
       .endSpec()
       .build()
@@ -534,11 +606,6 @@ private[spark] class KubernetesClusterSchedulerBackend(
     }
   }
 
-  override def createDriverEndpoint(
-    properties: Seq[(String, String)]): DriverEndpoint = {
-    new KubernetesDriverEndpoint(rpcEnv, properties)
-  }
-
   override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = Future[Boolean] {
     totalExpectedExecutors.set(requestedTotal)
     true
@@ -547,8 +614,10 @@ private[spark] class KubernetesClusterSchedulerBackend(
   override def doKillExecutors(executorIds: Seq[String]): Future[Boolean] = Future[Boolean] {
     RUNNING_EXECUTOR_PODS_LOCK.synchronized {
       for (executor <- executorIds) {
-        runningExecutorPods.remove(executor) match {
-          case Some(pod) => kubernetesClient.pods().delete(pod)
+        runningExecutorsToPods.remove(executor) match {
+          case Some(pod) =>
+            kubernetesClient.pods().delete(pod)
+            runningPodsToExecutors.remove(pod.getMetadata.getName)
           case None => logWarning(s"Unable to remove pod for unknown executor $executor")
         }
       }
@@ -564,6 +633,8 @@ private[spark] class KubernetesClusterSchedulerBackend(
 
   private class ExecutorPodsWatcher extends Watcher[Pod] {
 
+    private val DEFAULT_CONTAINER_FAILURE_EXIT_STATUS = -1
+
     override def eventReceived(action: Action, pod: Pod): Unit = {
       if (action == Action.MODIFIED && pod.getStatus.getPhase == "Running"
           && pod.getMetadata.getDeletionTimestamp == null) {
@@ -583,12 +654,75 @@ private[spark] class KubernetesClusterSchedulerBackend(
             executorPodsByIPs -= podIP
           }
         }
+        if (action == Action.ERROR) {
+          logInfo(s"Received pod $podName exited event. Reason: " + pod.getStatus.getReason)
+          handleErroredPod(pod)
+        } else if (action == Action.DELETED) {
+          logInfo(s"Received delete pod $podName event. Reason: " + pod.getStatus.getReason)
+          handleDeletedPod(pod)
+        }
       }
     }
 
     override def onClose(cause: KubernetesClientException): Unit = {
       logDebug("Executor pod watch closed.", cause)
     }
+
+    def getExecutorExitStatus(pod: Pod): Int = {
+      val containerStatuses = pod.getStatus.getContainerStatuses
+      if (!containerStatuses.isEmpty) {
+        // we assume the first container represents the pod status. This assumption may not hold
+        // true in the future. Revisit this if side-car containers start running inside executor
+        // pods.
+        getExecutorExitStatus(containerStatuses.get(0))
+      } else DEFAULT_CONTAINER_FAILURE_EXIT_STATUS
+    }
+
+    def getExecutorExitStatus(containerStatus: ContainerStatus): Int = {
+      Option(containerStatus.getState).map(containerState =>
+        Option(containerState.getTerminated).map(containerStateTerminated =>
+          containerStateTerminated.getExitCode.intValue()).getOrElse(UNKNOWN_EXIT_CODE)
+      ).getOrElse(UNKNOWN_EXIT_CODE)
+    }
+
+    def isPodAlreadyReleased(pod: Pod): Boolean = {
+      RUNNING_EXECUTOR_PODS_LOCK.synchronized {
+        !runningPodsToExecutors.contains(pod.getMetadata.getName)
+      }
+    }
+
+    def handleErroredPod(pod: Pod): Unit = {
+      val alreadyReleased = isPodAlreadyReleased(pod)
+      val containerExitStatus = getExecutorExitStatus(pod)
+      // container was probably actively killed by the driver.
+      val exitReason = if (alreadyReleased) {
+          ExecutorExited(containerExitStatus, exitCausedByApp = false,
+            s"Container in pod " + pod.getMetadata.getName +
+              " exited from explicit termination request.")
+        } else {
+          val containerExitReason = containerExitStatus match {
+            case VMEM_EXCEEDED_EXIT_CODE | PMEM_EXCEEDED_EXIT_CODE =>
+              memLimitExceededLogMessage(pod.getStatus.getReason)
+            case _ =>
+              // Here we can't be sure that that exit was caused by the application but this seems
+              // to be the right default since we know the pod was not explicitly deleted by
+              // the user.
+              "Pod exited with following container exit status code " + containerExitStatus
+          }
+          ExecutorExited(containerExitStatus, exitCausedByApp = true, containerExitReason)
+        }
+        failedPods.put(pod.getMetadata.getName, exitReason)
+    }
+
+    def handleDeletedPod(pod: Pod): Unit = {
+      val exitReason = ExecutorExited(getExecutorExitStatus(pod), exitCausedByApp = false,
+        "Pod " + pod.getMetadata.getName + " deleted or lost.")
+        failedPods.put(pod.getMetadata.getName, exitReason)
+    }
+  }
+
+  override def createDriverEndpoint(properties: Seq[(String, String)]): DriverEndpoint = {
+    new KubernetesDriverEndpoint(rpcEnv, properties)
   }
 
   private class KubernetesDriverEndpoint(
@@ -597,7 +731,16 @@ private[spark] class KubernetesClusterSchedulerBackend(
     extends DriverEndpoint(rpcEnv, sparkProperties) {
     private val externalShufflePort = conf.getInt("spark.shuffle.service.port", 7337)
 
-    override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
+    override def onDisconnected(rpcAddress: RpcAddress): Unit = {
+      addressToExecutorId.get(rpcAddress).foreach { executorId =>
+        if (disableExecutor(executorId)) {
+            executorsToRemove.add(executorId)
+        }
+      }
+    }
+
+    override def receiveAndReply(
+      context: RpcCallContext): PartialFunction[Any, Unit] = {
       new PartialFunction[Any, Unit]() {
         override def isDefinedAt(msg: Any): Boolean = {
           msg match {
@@ -614,7 +757,7 @@ private[spark] class KubernetesClusterSchedulerBackend(
                 var resolvedProperties = sparkProperties
                 val runningExecutorPod = kubernetesClient
                   .pods()
-                  .withName(runningExecutorPods(executorId).getMetadata.getName)
+                  .withName(runningExecutorsToPods(executorId).getMetadata.getName)
                   .get()
                 val nodeName = runningExecutorPod.getSpec.getNodeName
                 val shufflePodIp = shufflePodCache.get.getShufflePodForExecutor(nodeName)
@@ -636,7 +779,6 @@ private[spark] class KubernetesClusterSchedulerBackend(
       }.orElse(super.receiveAndReply(context))
     }
   }
-
 }
 case class ShuffleServiceConfig(
     shuffleNamespace: String,
@@ -646,6 +788,14 @@ case class ShuffleServiceConfig(
 private object KubernetesClusterSchedulerBackend {
   private val DEFAULT_STATIC_PORT = 10000
   private val EXECUTOR_ID_COUNTER = new AtomicLong(0L)
+  private val VMEM_EXCEEDED_EXIT_CODE = -103
+  private val PMEM_EXCEEDED_EXIT_CODE = -104
+  private val UNKNOWN_EXIT_CODE = -111
+
+  def memLimitExceededLogMessage(diagnostics: String): String = {
+    s"Pod/Container killed for exceeding memory limits. $diagnostics" +
+      " Consider boosting spark executor memory overhead."
+  }
 }
 
 /**
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSchedulerImpl.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSchedulerImpl.scala
index a5e126480b83d..ca40d46400ea2 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSchedulerImpl.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSchedulerImpl.scala
@@ -16,12 +16,59 @@
  */
 package org.apache.spark.scheduler.cluster.kubernetes
 
+import org.apache.spark.deploy.kubernetes.config._
+import org.apache.spark.scheduler.{SchedulerBackend, TaskSchedulerImpl, TaskSet, TaskSetManager}
+import org.apache.spark.util.Utils
 import org.apache.spark.SparkContext
-import org.apache.spark.scheduler.{TaskSchedulerImpl, TaskSet, TaskSetManager}
 
-private[spark] class KubernetesTaskSchedulerImpl(sc: SparkContext) extends TaskSchedulerImpl(sc) {
+private[spark] class KubernetesTaskSchedulerImpl(
+    sc: SparkContext,
+    rackResolverUtil: RackResolverUtil,
+    inetAddressUtil: InetAddressUtil = InetAddressUtilImpl) extends TaskSchedulerImpl(sc) {
 
+  var kubernetesSchedulerBackend: KubernetesClusterSchedulerBackend = null
+
+  def this(sc: SparkContext) = this(sc, new RackResolverUtilImpl(sc.hadoopConfiguration))
+
+  override def initialize(backend: SchedulerBackend): Unit = {
+    super.initialize(backend)
+    kubernetesSchedulerBackend = this.backend.asInstanceOf[KubernetesClusterSchedulerBackend]
+  }
   override def createTaskSetManager(taskSet: TaskSet, maxTaskFailures: Int): TaskSetManager = {
     new KubernetesTaskSetManager(this, taskSet, maxTaskFailures)
   }
+
+  override def getRackForHost(hostPort: String): Option[String] = {
+    if (!rackResolverUtil.isConfigured) {
+      // Only calls resolver when it is configured to avoid sending DNS queries for cluster nodes.
+      // See InetAddressUtil for details.
+      None
+    } else {
+      getRackForDatanodeOrExecutor(hostPort)
+    }
+  }
+
+  private def getRackForDatanodeOrExecutor(hostPort: String): Option[String] = {
+    val host = Utils.parseHostPort(hostPort)._1
+    val executorPod = kubernetesSchedulerBackend.getExecutorPodByIP(host)
+    val hadoopConfiguration = sc.hadoopConfiguration
+    executorPod.map(
+        pod => {
+          val clusterNodeName = pod.getSpec.getNodeName
+          val rackByNodeName = rackResolverUtil.resolveRack(hadoopConfiguration, clusterNodeName)
+          rackByNodeName.orElse({
+            val clusterNodeIP = pod.getStatus.getHostIP
+            val rackByNodeIP = rackResolverUtil.resolveRack(hadoopConfiguration, clusterNodeIP)
+            rackByNodeIP.orElse({
+              if (conf.get(KUBERNETES_DRIVER_CLUSTER_NODENAME_DNS_LOOKUP_ENABLED)) {
+                val clusterNodeFullName = inetAddressUtil.getFullHostName(clusterNodeIP)
+                rackResolverUtil.resolveRack(hadoopConfiguration, clusterNodeFullName)
+              } else {
+                Option.empty
+              }
+            })
+          })
+        }
+      ).getOrElse(rackResolverUtil.resolveRack(hadoopConfiguration, host))
+  }
 }
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSetManager.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSetManager.scala
index 51566d03a7a6c..44d01a5d76b01 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSetManager.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSetManager.scala
@@ -16,19 +16,20 @@
  */
 package org.apache.spark.scheduler.cluster.kubernetes
 
-import java.net.InetAddress
-
 import scala.collection.mutable.ArrayBuffer
 
+import org.apache.spark.deploy.kubernetes.config._
 import org.apache.spark.scheduler.{TaskSchedulerImpl, TaskSet, TaskSetManager}
 
 private[spark] class KubernetesTaskSetManager(
     sched: TaskSchedulerImpl,
     taskSet: TaskSet,
     maxTaskFailures: Int,
-    inetAddressUtil: InetAddressUtil = new InetAddressUtil)
+    inetAddressUtil: InetAddressUtil = InetAddressUtilImpl)
   extends TaskSetManager(sched, taskSet, maxTaskFailures) {
 
+  private val conf = sched.sc.conf
+
   /**
    * Overrides the lookup to use not only the executor pod IP, but also the cluster node
    * name and host IP address that the pod is running on. The base class may have populated
@@ -58,13 +59,19 @@ private[spark] class KubernetesTaskSetManager(
               s"$executorIP using cluster node IP $clusterNodeIP")
             pendingTasksClusterNodeIP
           } else {
-            val clusterNodeFullName = inetAddressUtil.getFullHostName(clusterNodeIP)
-            val pendingTasksClusterNodeFullName = super.getPendingTasksForHost(clusterNodeFullName)
-            if (pendingTasksClusterNodeFullName.nonEmpty) {
-              logDebug(s"Got preferred task list $pendingTasksClusterNodeFullName " +
-                s"for executor host $executorIP using cluster node full name $clusterNodeFullName")
+            if (conf.get(KUBERNETES_DRIVER_CLUSTER_NODENAME_DNS_LOOKUP_ENABLED)) {
+              val clusterNodeFullName = inetAddressUtil.getFullHostName(clusterNodeIP)
+              val pendingTasksClusterNodeFullName = super.getPendingTasksForHost(
+                clusterNodeFullName)
+              if (pendingTasksClusterNodeFullName.nonEmpty) {
+                logDebug(s"Got preferred task list $pendingTasksClusterNodeFullName " +
+                  s"for executor host $executorIP using cluster node full name " +
+                  s"$clusterNodeFullName")
+              }
+              pendingTasksClusterNodeFullName
+            } else {
+              pendingTasksExecutorIP  // Empty
             }
-            pendingTasksClusterNodeFullName
           }
         }
       } else {
@@ -74,12 +81,3 @@ private[spark] class KubernetesTaskSetManager(
   }
 }
 
-// To support mocks in unit tests.
-private[kubernetes] class InetAddressUtil {
-
-  // NOTE: This does issue a network call to DNS. Caching is done internally by the InetAddress
-  // class for both hits and misses.
-  def getFullHostName(ipAddress: String): String = {
-    InetAddress.getByName(ipAddress).getCanonicalHostName
-  }
-}
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/RackResolverUtil.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/RackResolverUtil.scala
new file mode 100644
index 0000000000000..29a7dc982a5af
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/RackResolverUtil.scala
@@ -0,0 +1,69 @@
+/*
+ * 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.scheduler.cluster.kubernetes
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic
+import org.apache.hadoop.net.{NetworkTopology, ScriptBasedMapping, TableMapping}
+import org.apache.hadoop.yarn.util.RackResolver
+import org.apache.log4j.{Level, Logger}
+
+/**
+ * Finds rack names that cluster nodes belong to in order to support HDFS rack locality.
+ */
+private[kubernetes] trait RackResolverUtil {
+
+  def isConfigured() : Boolean
+
+  def resolveRack(hadoopConfiguration: Configuration, host: String): Option[String]
+}
+
+private[kubernetes] class RackResolverUtilImpl(hadoopConfiguration: Configuration)
+    extends RackResolverUtil {
+
+  val scriptPlugin : String = classOf[ScriptBasedMapping].getCanonicalName
+  val tablePlugin : String = classOf[TableMapping].getCanonicalName
+  val isResolverConfigured : Boolean = checkConfigured(hadoopConfiguration)
+
+  // RackResolver logs an INFO message whenever it resolves a rack, which is way too often.
+  if (Logger.getLogger(classOf[RackResolver]).getLevel == null) {
+    Logger.getLogger(classOf[RackResolver]).setLevel(Level.WARN)
+  }
+
+  override def isConfigured() : Boolean = isResolverConfigured
+
+  def checkConfigured(hadoopConfiguration: Configuration): Boolean = {
+    val plugin = hadoopConfiguration.get(
+      CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY, scriptPlugin)
+    val scriptName = hadoopConfiguration.get(
+      CommonConfigurationKeysPublic.NET_TOPOLOGY_SCRIPT_FILE_NAME_KEY, "")
+    val tableName = hadoopConfiguration.get(
+      CommonConfigurationKeysPublic.NET_TOPOLOGY_TABLE_MAPPING_FILE_KEY, "")
+    plugin == scriptPlugin && scriptName.nonEmpty ||
+      plugin == tablePlugin && tableName.nonEmpty ||
+      plugin != scriptPlugin && plugin != tablePlugin
+  }
+
+  override def resolveRack(hadoopConfiguration: Configuration, host: String): Option[String] = {
+    val rack = Option(RackResolver.resolve(hadoopConfiguration, host).getNetworkLocation)
+    if (rack.nonEmpty && rack.get != NetworkTopology.DEFAULT_RACK) {
+      rack
+    } else {
+      None
+    }
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/ClientSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/ClientSuite.scala
index 965ee75c248b8..0100dce454a3f 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/ClientSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/ClientSuite.scala
@@ -135,7 +135,7 @@ class ClientSuite extends SparkFunSuite with BeforeAndAfter {
         .set("spark.logConf", "true")
         .set(
           org.apache.spark.internal.config.DRIVER_JAVA_OPTIONS,
-          "-XX:+|-HeapDumpOnOutOfMemoryError")
+          "-XX:+HeapDumpOnOutOfMemoryError -XX:+PrintGCDetails")
     val submissionClient = new Client(
         submissionSteps,
         sparkConf,
@@ -147,15 +147,22 @@ class ClientSuite extends SparkFunSuite with BeforeAndAfter {
     val createdPod = createdPodArgumentCaptor.getValue
     val driverContainer = Iterables.getOnlyElement(createdPod.getSpec.getContainers)
     assert(driverContainer.getName === SecondTestConfigurationStep.containerName)
-    val driverJvmOptsEnv = Iterables.getOnlyElement(driverContainer.getEnv)
-    assert(driverJvmOptsEnv.getName === ENV_DRIVER_JAVA_OPTS)
-    val driverJvmOpts = driverJvmOptsEnv.getValue.split(" ").toSet
-    assert(driverJvmOpts.contains("-Dspark.logConf=true"))
-    assert(driverJvmOpts.contains(
+    val driverJvmOptsEnvs = driverContainer.getEnv.asScala.filter { env =>
+      env.getName.startsWith(ENV_JAVA_OPT_PREFIX)
+    }.sortBy(_.getName)
+    assert(driverJvmOptsEnvs.size === 4)
+
+    val expectedJvmOptsValues = Seq(
+        "-Dspark.logConf=true",
         s"-D${SecondTestConfigurationStep.sparkConfKey}=" +
-          SecondTestConfigurationStep.sparkConfValue))
-    assert(driverJvmOpts.contains(
-        "-XX:+|-HeapDumpOnOutOfMemoryError"))
+            s"${SecondTestConfigurationStep.sparkConfValue}",
+        s"-XX:+HeapDumpOnOutOfMemoryError",
+        s"-XX:+PrintGCDetails")
+    driverJvmOptsEnvs.zip(expectedJvmOptsValues).zipWithIndex.foreach {
+      case ((resolvedEnv, expectedJvmOpt), index) =>
+        assert(resolvedEnv.getName === s"$ENV_JAVA_OPT_PREFIX$index")
+        assert(resolvedEnv.getValue === expectedJvmOpt)
+    }
   }
 
   test("Waiting for app completion should stall on the watcher") {
@@ -211,8 +218,8 @@ private object SecondTestConfigurationStep extends DriverConfigurationStep {
   override def configureDriver(driverSpec: KubernetesDriverSpec): KubernetesDriverSpec = {
     val modifiedPod = new PodBuilder(driverSpec.driverPod)
       .editMetadata()
-      .addToAnnotations(annotationKey, annotationValue)
-      .endMetadata()
+        .addToAnnotations(annotationKey, annotationValue)
+        .endMetadata()
       .build()
     val resolvedSparkConf = driverSpec.driverSparkConf.clone().set(sparkConfKey, sparkConfValue)
     val modifiedContainer = new ContainerBuilder(driverSpec.driverContainer)
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/BaseDriverConfigurationStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/BaseDriverConfigurationStepSuite.scala
index c7d80a16a1532..4520c40ec81c1 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/BaseDriverConfigurationStepSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/BaseDriverConfigurationStepSuite.scala
@@ -36,6 +36,8 @@ private[spark] class BaseDriverConfigurationStepSuite extends SparkFunSuite {
   private val CUSTOM_ANNOTATION_VALUE = "customAnnotationValue"
   private val DEPRECATED_CUSTOM_ANNOTATION_KEY = "customAnnotationDeprecated"
   private val DEPRECATED_CUSTOM_ANNOTATION_VALUE = "customAnnotationDeprecatedValue"
+  private val DRIVER_CUSTOM_ENV_KEY1 = "customDriverEnv1"
+  private val DRIVER_CUSTOM_ENV_KEY2 = "customDriverEnv2"
 
   test("Set all possible configurations from the user.") {
     val sparkConf = new SparkConf()
@@ -49,6 +51,9 @@ private[spark] class BaseDriverConfigurationStepSuite extends SparkFunSuite {
         .set(s"spark.kubernetes.driver.annotation.$CUSTOM_ANNOTATION_KEY", CUSTOM_ANNOTATION_VALUE)
         .set("spark.kubernetes.driver.annotations",
             s"$DEPRECATED_CUSTOM_ANNOTATION_KEY=$DEPRECATED_CUSTOM_ANNOTATION_VALUE")
+        .set(s"$KUBERNETES_DRIVER_ENV_KEY$DRIVER_CUSTOM_ENV_KEY1", "customDriverEnv1")
+        .set(s"$KUBERNETES_DRIVER_ENV_KEY$DRIVER_CUSTOM_ENV_KEY2", "customDriverEnv2")
+
     val submissionStep = new BaseDriverConfigurationStep(
         APP_ID,
         RESOURCE_NAME_PREFIX,
@@ -74,11 +79,13 @@ private[spark] class BaseDriverConfigurationStepSuite extends SparkFunSuite {
         .asScala
         .map(env => (env.getName, env.getValue))
         .toMap
-    assert(envs.size === 4)
+    assert(envs.size === 6)
     assert(envs(ENV_SUBMIT_EXTRA_CLASSPATH) === "/opt/spark/spark-exmaples.jar")
     assert(envs(ENV_DRIVER_MEMORY) === "456m")
     assert(envs(ENV_DRIVER_MAIN_CLASS) === MAIN_CLASS)
     assert(envs(ENV_DRIVER_ARGS) === "arg1 arg2")
+    assert(envs(DRIVER_CUSTOM_ENV_KEY1) === "customDriverEnv1")
+    assert(envs(DRIVER_CUSTOM_ENV_KEY2) === "customDriverEnv2")
     val resourceRequirements = preparedDriverSpec.driverContainer.getResources
     val requests = resourceRequirements.getRequests.asScala
     assert(requests("cpu").getAmount === "2")
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSchedulerImplSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSchedulerImplSuite.scala
new file mode 100644
index 0000000000000..e10b9508530f7
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSchedulerImplSuite.scala
@@ -0,0 +1,164 @@
+/*
+ * 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.scheduler.cluster.kubernetes
+
+import io.fabric8.kubernetes.api.model.{Pod, PodSpec, PodStatus}
+import org.mockito.Matchers._
+import org.mockito.Mockito._
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.{SparkContext, SparkFunSuite}
+import org.apache.spark.deploy.kubernetes.config._
+import org.apache.spark.scheduler.FakeTask
+
+class KubernetesTaskSchedulerImplSuite extends SparkFunSuite with BeforeAndAfter {
+
+  SparkContext.clearActiveContext()
+  val sc = new SparkContext("local", "test")
+  val backend = mock(classOf[KubernetesClusterSchedulerBackend])
+
+  before {
+    sc.conf.remove(KUBERNETES_DRIVER_CLUSTER_NODENAME_DNS_LOOKUP_ENABLED)
+  }
+
+  test("Create a k8s task set manager") {
+    val sched = new KubernetesTaskSchedulerImpl(sc)
+    sched.kubernetesSchedulerBackend = backend
+    val taskSet = FakeTask.createTaskSet(0)
+
+    val manager = sched.createTaskSetManager(taskSet, maxTaskFailures = 3)
+    assert(manager.isInstanceOf[KubernetesTaskSetManager])
+  }
+
+  test("Gets racks for datanodes") {
+    val rackResolverUtil = mock(classOf[RackResolverUtil])
+    when(rackResolverUtil.isConfigured).thenReturn(true)
+    when(rackResolverUtil.resolveRack(sc.hadoopConfiguration, "kube-node1"))
+      .thenReturn(Option("/rack1"))
+    when(rackResolverUtil.resolveRack(sc.hadoopConfiguration, "kube-node2"))
+      .thenReturn(Option("/rack2"))
+    val sched = new KubernetesTaskSchedulerImpl(sc, rackResolverUtil)
+    sched.kubernetesSchedulerBackend = backend
+    when(backend.getExecutorPodByIP("kube-node1")).thenReturn(None)
+    when(backend.getExecutorPodByIP("kube-node2")).thenReturn(None)
+
+    assert(sched.getRackForHost("kube-node1:60010") == Option("/rack1"))
+    assert(sched.getRackForHost("kube-node2:60010") == Option("/rack2"))
+  }
+
+  test("Gets racks for executor pods") {
+    sc.conf.set(KUBERNETES_DRIVER_CLUSTER_NODENAME_DNS_LOOKUP_ENABLED, true)
+    val rackResolverUtil = mock(classOf[RackResolverUtil])
+    when(rackResolverUtil.isConfigured).thenReturn(true)
+    when(rackResolverUtil.resolveRack(sc.hadoopConfiguration, "kube-node1"))
+      .thenReturn(Option("/rack1"))
+    when(rackResolverUtil.resolveRack(sc.hadoopConfiguration, "kube-node2.mydomain.com"))
+      .thenReturn(Option("/rack2"))
+    when(rackResolverUtil.resolveRack(sc.hadoopConfiguration, "kube-node2"))
+      .thenReturn(None)
+    when(rackResolverUtil.resolveRack(sc.hadoopConfiguration, "192.168.1.5"))
+      .thenReturn(None)
+    val inetAddressUtil = mock(classOf[InetAddressUtil])
+    val sched = new KubernetesTaskSchedulerImpl(sc, rackResolverUtil, inetAddressUtil)
+    sched.kubernetesSchedulerBackend = backend
+
+    val spec1 = mock(classOf[PodSpec])
+    when(spec1.getNodeName).thenReturn("kube-node1")
+    val status1 = mock(classOf[PodStatus])
+    when(status1.getHostIP).thenReturn("192.168.1.4")
+    val pod1 = mock(classOf[Pod])
+    when(pod1.getSpec).thenReturn(spec1)
+    when(pod1.getStatus).thenReturn(status1)
+    when(backend.getExecutorPodByIP("10.0.0.1")).thenReturn(Some(pod1))
+
+    val spec2 = mock(classOf[PodSpec])
+    when(spec2.getNodeName).thenReturn("kube-node2")
+    val status2 = mock(classOf[PodStatus])
+    when(status2.getHostIP).thenReturn("192.168.1.5")
+    val pod2 = mock(classOf[Pod])
+    when(pod2.getSpec).thenReturn(spec2)
+    when(pod2.getStatus).thenReturn(status2)
+    when(inetAddressUtil.getFullHostName("192.168.1.5")).thenReturn("kube-node2.mydomain.com")
+    when(backend.getExecutorPodByIP("10.0.1.1")).thenReturn(Some(pod2))
+
+    assert(sched.getRackForHost("10.0.0.1:7079") == Option("/rack1"))
+    assert(sched.getRackForHost("10.0.1.1:7079") == Option("/rack2"))
+
+    verify(inetAddressUtil, times(1)).getFullHostName(anyString())
+  }
+
+  test("Gets racks for executor pods while disabling DNS lookup ") {
+    sc.conf.set(KUBERNETES_DRIVER_CLUSTER_NODENAME_DNS_LOOKUP_ENABLED, false)
+    val rackResolverUtil = mock(classOf[RackResolverUtil])
+    when(rackResolverUtil.isConfigured).thenReturn(true)
+    when(rackResolverUtil.resolveRack(sc.hadoopConfiguration, "kube-node1"))
+      .thenReturn(Option("/rack1"))
+    when(rackResolverUtil.resolveRack(sc.hadoopConfiguration, "kube-node2.mydomain.com"))
+      .thenReturn(Option("/rack2"))
+    when(rackResolverUtil.resolveRack(sc.hadoopConfiguration, "kube-node2"))
+      .thenReturn(None)
+    when(rackResolverUtil.resolveRack(sc.hadoopConfiguration, "192.168.1.5"))
+      .thenReturn(None)
+    val inetAddressUtil = mock(classOf[InetAddressUtil])
+    val sched = new KubernetesTaskSchedulerImpl(sc, rackResolverUtil, inetAddressUtil)
+    sched.kubernetesSchedulerBackend = backend
+
+    val spec1 = mock(classOf[PodSpec])
+    when(spec1.getNodeName).thenReturn("kube-node1")
+    val status1 = mock(classOf[PodStatus])
+    when(status1.getHostIP).thenReturn("192.168.1.4")
+    val pod1 = mock(classOf[Pod])
+    when(pod1.getSpec).thenReturn(spec1)
+    when(pod1.getStatus).thenReturn(status1)
+    when(backend.getExecutorPodByIP("10.0.0.1")).thenReturn(Some(pod1))
+
+    val spec2 = mock(classOf[PodSpec])
+    when(spec2.getNodeName).thenReturn("kube-node2")
+    val status2 = mock(classOf[PodStatus])
+    when(status2.getHostIP).thenReturn("192.168.1.5")
+    val pod2 = mock(classOf[Pod])
+    when(pod2.getSpec).thenReturn(spec2)
+    when(pod2.getStatus).thenReturn(status2)
+    when(inetAddressUtil.getFullHostName("192.168.1.5")).thenReturn("kube-node2.mydomain.com")
+    when(backend.getExecutorPodByIP("10.0.1.1")).thenReturn(Some(pod2))
+
+    assert(sched.getRackForHost("10.0.0.1:7079") == Option("/rack1"))
+    assert(sched.getRackForHost("10.0.1.1:7079") == None)
+
+    verify(inetAddressUtil, never).getFullHostName(anyString())
+  }
+
+  test("Does not get racks if plugin is not configured") {
+    val rackResolverUtil = mock(classOf[RackResolverUtil])
+    when(rackResolverUtil.isConfigured()).thenReturn(false)
+    val sched = new KubernetesTaskSchedulerImpl(sc, rackResolverUtil)
+    sched.kubernetesSchedulerBackend = backend
+    when(backend.getExecutorPodByIP("kube-node1")).thenReturn(None)
+
+    val spec1 = mock(classOf[PodSpec])
+    when(spec1.getNodeName).thenReturn("kube-node1")
+    val status1 = mock(classOf[PodStatus])
+    when(status1.getHostIP).thenReturn("192.168.1.4")
+    val pod1 = mock(classOf[Pod])
+    when(pod1.getSpec).thenReturn(spec1)
+    when(pod1.getStatus).thenReturn(status1)
+    when(backend.getExecutorPodByIP("10.0.0.1")).thenReturn(Some(pod1))
+
+    assert(sched.getRackForHost("kube-node1:60010").isEmpty)
+    assert(sched.getRackForHost("10.0.0.1:7079").isEmpty)
+  }
+}
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSetManagerSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSetManagerSuite.scala
index 7618c137ab22b..889758731a6c1 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSetManagerSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesTaskSetManagerSuite.scala
@@ -20,18 +20,25 @@ import scala.collection.mutable.ArrayBuffer
 
 import io.fabric8.kubernetes.api.model.{Pod, PodSpec, PodStatus}
 import org.mockito.Mockito._
+import org.scalatest.BeforeAndAfter
 
 import org.apache.spark.{SparkContext, SparkFunSuite}
+import org.apache.spark.deploy.kubernetes.config._
 import org.apache.spark.scheduler.{FakeTask, FakeTaskScheduler, HostTaskLocation, TaskLocation}
 
-class KubernetesTaskSetManagerSuite extends SparkFunSuite {
+class KubernetesTaskSetManagerSuite extends SparkFunSuite with BeforeAndAfter {
 
+  SparkContext.clearActiveContext()
   val sc = new SparkContext("local", "test")
   val sched = new FakeTaskScheduler(sc,
     ("execA", "10.0.0.1"), ("execB", "10.0.0.2"), ("execC", "10.0.0.3"))
   val backend = mock(classOf[KubernetesClusterSchedulerBackend])
   sched.backend = backend
 
+  before {
+    sc.conf.remove(KUBERNETES_DRIVER_CLUSTER_NODENAME_DNS_LOOKUP_ENABLED)
+  }
+
   test("Find pending tasks for executors using executor pod IP addresses") {
     val taskSet = FakeTask.createTaskSet(3,
       Seq(TaskLocation("10.0.0.1", "execA")),  // Task 0 runs on executor pod 10.0.0.1.
@@ -76,7 +83,33 @@ class KubernetesTaskSetManagerSuite extends SparkFunSuite {
     assert(manager.getPendingTasksForHost("10.0.0.1") == ArrayBuffer(1, 0))
   }
 
+  test("Test DNS lookup is disabled by default for cluster node full hostnames") {
+    assert(!sc.conf.get(KUBERNETES_DRIVER_CLUSTER_NODENAME_DNS_LOOKUP_ENABLED))
+  }
+
+  test("Find pending tasks for executors, but avoid looking up cluster node FQDNs from DNS") {
+    sc.conf.set(KUBERNETES_DRIVER_CLUSTER_NODENAME_DNS_LOOKUP_ENABLED, false)
+    val taskSet = FakeTask.createTaskSet(2,
+      Seq(HostTaskLocation("kube-node1.domain1")),  // Task 0's partition belongs to datanode here.
+      Seq(HostTaskLocation("kube-node1.domain1"))   // task 1's partition belongs to datanode here.
+    )
+    val spec1 = mock(classOf[PodSpec])
+    when(spec1.getNodeName).thenReturn("kube-node1")
+    val pod1 = mock(classOf[Pod])
+    when(pod1.getSpec).thenReturn(spec1)
+    val status1 = mock(classOf[PodStatus])
+    when(status1.getHostIP).thenReturn("196.0.0.5")
+    when(pod1.getStatus).thenReturn(status1)
+    val inetAddressUtil = mock(classOf[InetAddressUtil])
+    when(inetAddressUtil.getFullHostName("196.0.0.5")).thenReturn("kube-node1.domain1")
+    when(backend.getExecutorPodByIP("10.0.0.1")).thenReturn(Some(pod1))
+
+    val manager = new KubernetesTaskSetManager(sched, taskSet, maxTaskFailures = 2, inetAddressUtil)
+    assert(manager.getPendingTasksForHost("10.0.0.1") == ArrayBuffer())
+  }
+
   test("Find pending tasks for executors using cluster node FQDNs that executor pods run on") {
+    sc.conf.set(KUBERNETES_DRIVER_CLUSTER_NODENAME_DNS_LOOKUP_ENABLED, true)
     val taskSet = FakeTask.createTaskSet(2,
       Seq(HostTaskLocation("kube-node1.domain1")),  // Task 0's partition belongs to datanode here.
       Seq(HostTaskLocation("kube-node1.domain1"))   // task 1's partition belongs to datanode here.
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/kubernetes/RackResolverUtilImplSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/kubernetes/RackResolverUtilImplSuite.scala
new file mode 100644
index 0000000000000..ee671ce7d6590
--- /dev/null
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/kubernetes/RackResolverUtilImplSuite.scala
@@ -0,0 +1,49 @@
+/*
+ * 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.scheduler.cluster.kubernetes
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic
+
+import org.apache.spark.SparkFunSuite
+
+class RackResolverUtilImplSuite extends SparkFunSuite {
+
+  test("Detects if topology plugin is configured") {
+    val hadoopConfiguration = new Configuration
+    val rackResolverUtil = new RackResolverUtilImpl(hadoopConfiguration)
+
+    assert(!rackResolverUtil.checkConfigured(hadoopConfiguration))
+    hadoopConfiguration.set(CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+      rackResolverUtil.scriptPlugin)
+    assert(!rackResolverUtil.checkConfigured(hadoopConfiguration))
+    hadoopConfiguration.set(CommonConfigurationKeysPublic.NET_TOPOLOGY_SCRIPT_FILE_NAME_KEY,
+      "my-script")
+    assert(rackResolverUtil.checkConfigured(hadoopConfiguration))
+
+    hadoopConfiguration.set(CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+      rackResolverUtil.tablePlugin)
+    assert(!rackResolverUtil.checkConfigured(hadoopConfiguration))
+    hadoopConfiguration.set(CommonConfigurationKeysPublic.NET_TOPOLOGY_TABLE_MAPPING_FILE_KEY,
+      "my-table")
+    assert(rackResolverUtil.checkConfigured(hadoopConfiguration))
+
+    hadoopConfiguration.set(CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+      "my.Plugin")
+    assert(rackResolverUtil.checkConfigured(hadoopConfiguration))
+  }
+}
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/assembly/docker-assembly.xml b/resource-managers/kubernetes/docker-minimal-bundle/src/main/assembly/docker-assembly.xml
index 2b48d366256fe..e6de2c11a0ecf 100644
--- a/resource-managers/kubernetes/docker-minimal-bundle/src/main/assembly/docker-assembly.xml
+++ b/resource-managers/kubernetes/docker-minimal-bundle/src/main/assembly/docker-assembly.xml
@@ -69,6 +69,7 @@
       <excludes>
         <exclude>org.apache.spark:spark-assembly_${scala.binary.version}:pom</exclude>
         <exclude>org.spark-project.spark:unused</exclude>
+        <exclude>com.sun.jersey:*</exclude>
       </excludes>
     </dependencySet>
   </dependencySets>
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/driver-py/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/driver-py/Dockerfile
index 6dcc7511c0dd9..41e4b31446c59 100644
--- a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/driver-py/Dockerfile
+++ b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/driver-py/Dockerfile
@@ -39,10 +39,10 @@ ENV PYSPARK_DRIVER_PYTHON python
 ENV PYTHONPATH ${SPARK_HOME}/python/:${SPARK_HOME}/python/lib/py4j-0.10.4-src.zip:${PYTHONPATH}
 
 CMD SPARK_CLASSPATH="${SPARK_HOME}/jars/*" && \
+    env | grep SPARK_JAVA_OPT_ | sed 's/[^=]*=\(.*\)/\1/g' > /tmp/java_opts.txt && \
+    readarray -t SPARK_DRIVER_JAVA_OPTS < /tmp/java_opts.txt && \
     if ! [ -z ${SPARK_MOUNTED_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_MOUNTED_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_SUBMIT_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_SUBMIT_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_MOUNTED_FILES_DIR} ]; then cp -R "$SPARK_MOUNTED_FILES_DIR/." .; fi && \
-    exec /sbin/tini -- ${JAVA_HOME}/bin/java $SPARK_DRIVER_JAVA_OPTS -cp $SPARK_CLASSPATH \
-    -Xms$SPARK_DRIVER_MEMORY -Xmx$SPARK_DRIVER_MEMORY \
-    $SPARK_DRIVER_CLASS $PYSPARK_PRIMARY $PYSPARK_FILES $SPARK_DRIVER_ARGS
+    ${JAVA_HOME}/bin/java "${SPARK_DRIVER_JAVA_OPTS[@]}" -cp $SPARK_CLASSPATH -Xms$SPARK_DRIVER_MEMORY -Xmx$SPARK_DRIVER_MEMORY $SPARK_DRIVER_CLASS $PYSPARK_PRIMARY $PYSPARK_FILES $SPARK_DRIVER_ARGS
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/driver/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/driver/Dockerfile
index 6bbff8ef64a0f..c2ff8f4f55822 100644
--- a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/driver/Dockerfile
+++ b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/driver/Dockerfile
@@ -24,8 +24,10 @@ FROM spark-base
 COPY examples /opt/spark/examples
 
 CMD SPARK_CLASSPATH="${SPARK_HOME}/jars/*" && \
+    env | grep SPARK_JAVA_OPT_ | sed 's/[^=]*=\(.*\)/\1/g' > /tmp/java_opts.txt && \
+    readarray -t SPARK_DRIVER_JAVA_OPTS < /tmp/java_opts.txt && \
     if ! [ -z ${SPARK_MOUNTED_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_MOUNTED_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_SUBMIT_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_SUBMIT_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \
-    if ! [ -z ${SPARK_MOUNTED_FILES_DIR} ]; then cp -R "$SPARK_MOUNTED_FILES_DIR/." .; fi && \
-    exec /sbin/tini -- ${JAVA_HOME}/bin/java $SPARK_DRIVER_JAVA_OPTS -cp $SPARK_CLASSPATH -Xms$SPARK_DRIVER_MEMORY -Xmx$SPARK_DRIVER_MEMORY $SPARK_DRIVER_CLASS $SPARK_DRIVER_ARGS
+    if ! [ -z ${SPARK_MOUNTED_FILES_DIR+x} ]; then cp -R "$SPARK_MOUNTED_FILES_DIR/." .; fi && \
+    ${JAVA_HOME}/bin/java "${SPARK_DRIVER_JAVA_OPTS[@]}" -cp $SPARK_CLASSPATH -Xms$SPARK_DRIVER_MEMORY -Xmx$SPARK_DRIVER_MEMORY $SPARK_DRIVER_CLASS $SPARK_DRIVER_ARGS
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor-py/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor-py/Dockerfile
index 7a65a4f879376..cecb0c598d618 100644
--- a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor-py/Dockerfile
+++ b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor-py/Dockerfile
@@ -38,9 +38,10 @@ ENV PYSPARK_PYTHON python
 ENV PYSPARK_DRIVER_PYTHON python
 ENV PYTHONPATH ${SPARK_HOME}/python/:${SPARK_HOME}/python/lib/py4j-0.10.4-src.zip:${PYTHONPATH}
 
-# TODO support spark.executor.extraClassPath
 CMD SPARK_CLASSPATH="${SPARK_HOME}/jars/*" && \
+    env | grep SPARK_JAVA_OPT_ | sed 's/[^=]*=\(.*\)/\1/g' > /tmp/java_opts.txt && \
+    readarray -t SPARK_EXECUTOR_JAVA_OPTS < /tmp/java_opts.txt && \
     if ! [ -z ${SPARK_MOUNTED_CLASSPATH}+x} ]; then SPARK_CLASSPATH="$SPARK_MOUNTED_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_EXECUTOR_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_EXECUTOR_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_MOUNTED_FILES_DIR} ]; then cp -R "$SPARK_MOUNTED_FILES_DIR/." .; fi && \
-    exec /sbin/tini -- ${JAVA_HOME}/bin/java -Dspark.executor.port=$SPARK_EXECUTOR_PORT -Xms$SPARK_EXECUTOR_MEMORY -Xmx$SPARK_EXECUTOR_MEMORY -cp $SPARK_CLASSPATH org.apache.spark.executor.CoarseGrainedExecutorBackend --driver-url $SPARK_DRIVER_URL --executor-id $SPARK_EXECUTOR_ID --cores $SPARK_EXECUTOR_CORES --app-id $SPARK_APPLICATION_ID --hostname $SPARK_EXECUTOR_POD_IP
\ No newline at end of file
+    ${JAVA_HOME}/bin/java "${SPARK_EXECUTOR_JAVA_OPTS[@]}" -Dspark.executor.port=$SPARK_EXECUTOR_PORT -Xms$SPARK_EXECUTOR_MEMORY -Xmx$SPARK_EXECUTOR_MEMORY -cp $SPARK_CLASSPATH org.apache.spark.executor.CoarseGrainedExecutorBackend --driver-url $SPARK_DRIVER_URL --executor-id $SPARK_EXECUTOR_ID --cores $SPARK_EXECUTOR_CORES --app-id $SPARK_APPLICATION_ID --hostname $SPARK_EXECUTOR_POD_IP
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor/Dockerfile
index 9c9efb23d7e95..437fb645a253d 100644
--- a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor/Dockerfile
+++ b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/executor/Dockerfile
@@ -23,9 +23,11 @@ FROM spark-base
 
 COPY examples /opt/spark/examples
 
-# TODO support spark.executor.extraClassPath
 CMD SPARK_CLASSPATH="${SPARK_HOME}/jars/*" && \
+    env | grep SPARK_JAVA_OPT_ | sed 's/[^=]*=\(.*\)/\1/g' > /tmp/java_opts.txt && \
+    readarray -t SPARK_EXECUTOR_JAVA_OPTS < /tmp/java_opts.txt && \
     if ! [ -z ${SPARK_MOUNTED_CLASSPATH}+x} ]; then SPARK_CLASSPATH="$SPARK_MOUNTED_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_EXECUTOR_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_EXECUTOR_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \
+    if ! [ -z ${SPARK_EXTRA_CLASSPATH+x} ]; then SPARK_CLASSPATH="$SPARK_EXTRA_CLASSPATH:$SPARK_CLASSPATH"; fi && \
     if ! [ -z ${SPARK_MOUNTED_FILES_DIR} ]; then cp -R "$SPARK_MOUNTED_FILES_DIR/." .; fi && \
-    exec /sbin/tini -- ${JAVA_HOME}/bin/java -Dspark.executor.port=$SPARK_EXECUTOR_PORT -Xms$SPARK_EXECUTOR_MEMORY -Xmx$SPARK_EXECUTOR_MEMORY -cp $SPARK_CLASSPATH org.apache.spark.executor.CoarseGrainedExecutorBackend --driver-url $SPARK_DRIVER_URL --executor-id $SPARK_EXECUTOR_ID --cores $SPARK_EXECUTOR_CORES --app-id $SPARK_APPLICATION_ID --hostname $SPARK_EXECUTOR_POD_IP
+    ${JAVA_HOME}/bin/java "${SPARK_EXECUTOR_JAVA_OPTS[@]}" -Dspark.executor.port=$SPARK_EXECUTOR_PORT -Xms$SPARK_EXECUTOR_MEMORY -Xmx$SPARK_EXECUTOR_MEMORY -cp $SPARK_CLASSPATH org.apache.spark.executor.CoarseGrainedExecutorBackend --driver-url $SPARK_DRIVER_URL --executor-id $SPARK_EXECUTOR_ID --cores $SPARK_EXECUTOR_CORES --app-id $SPARK_APPLICATION_ID --hostname $SPARK_EXECUTOR_POD_IP
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/init-container/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/init-container/Dockerfile
index 4bafe25e2608f..2ab3e6295b6d8 100644
--- a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/init-container/Dockerfile
+++ b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/init-container/Dockerfile
@@ -21,4 +21,4 @@ FROM spark-base
 # command should be invoked from the top level directory of the Spark distribution. E.g.:
 # docker build -t spark-init:latest -f dockerfiles/init-container/Dockerfile .
 
-ENTRYPOINT [ "/sbin/tini", "--", "bin/spark-class", "org.apache.spark.deploy.rest.kubernetes.KubernetesSparkDependencyDownloadInitContainer" ]
+ENTRYPOINT [ "/opt/entrypoint.sh", "bin/spark-class", "org.apache.spark.deploy.rest.kubernetes.KubernetesSparkDependencyDownloadInitContainer" ]
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/resource-staging-server/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/resource-staging-server/Dockerfile
index 9ca96be0f1a88..0e0c9dd31aad6 100644
--- a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/resource-staging-server/Dockerfile
+++ b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/resource-staging-server/Dockerfile
@@ -22,4 +22,4 @@ FROM spark-base
 # command should be invoked from the top level directory of the Spark distribution. E.g.:
 # docker build -t spark-resource-staging-server:latest -f dockerfiles/resource-staging-server/Dockerfile .
 
-ENTRYPOINT [ "/sbin/tini", "--", "bin/spark-class", "org.apache.spark.deploy.rest.kubernetes.ResourceStagingServer" ]
+ENTRYPOINT [ "/opt/entrypoint.sh", "bin/spark-class", "org.apache.spark.deploy.rest.kubernetes.ResourceStagingServer" ]
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/shuffle-service/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/shuffle-service/Dockerfile
index ccb2f1a03d88c..2ae0be4ee6c32 100644
--- a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/shuffle-service/Dockerfile
+++ b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/shuffle-service/Dockerfile
@@ -23,4 +23,4 @@ FROM spark-base
 
 COPY examples /opt/spark/examples
 
-ENTRYPOINT [ "/sbin/tini", "--", "bin/spark-class", "org.apache.spark.deploy.kubernetes.KubernetesExternalShuffleService", "1" ]
+ENTRYPOINT [ "/opt/entrypoint.sh", "bin/spark-class", "org.apache.spark.deploy.kubernetes.KubernetesExternalShuffleService", "1" ]
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/spark-base/Dockerfile b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/spark-base/Dockerfile
index b0925e3bb0416..a982fc5fd30ce 100644
--- a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/spark-base/Dockerfile
+++ b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/spark-base/Dockerfile
@@ -23,13 +23,19 @@ FROM openjdk:8-alpine
 RUN apk upgrade --no-cache && \
     apk add --no-cache bash tini && \
     mkdir -p /opt/spark && \
-    touch /opt/spark/RELEASE
+    touch /opt/spark/RELEASE && \
+    rm /bin/sh && \
+    ln -sv /bin/bash /bin/sh && \
+    chgrp root /etc/passwd && chmod ug+rw /etc/passwd
 
 COPY jars /opt/spark/jars
 COPY bin /opt/spark/bin
 COPY sbin /opt/spark/sbin
 COPY conf /opt/spark/conf
+COPY dockerfiles/spark-base/entrypoint.sh /opt/
 
 ENV SPARK_HOME /opt/spark
 
 WORKDIR /opt/spark
+
+ENTRYPOINT [ "/opt/entrypoint.sh" ]
diff --git a/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/spark-base/entrypoint.sh b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/spark-base/entrypoint.sh
new file mode 100755
index 0000000000000..12713160abefa
--- /dev/null
+++ b/resource-managers/kubernetes/docker-minimal-bundle/src/main/docker/spark-base/entrypoint.sh
@@ -0,0 +1,34 @@
+#!/bin/bash
+#
+# 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.
+#
+
+# Check whether there is a passwd entry for the container UID
+myuid=$(id -u)
+mygid=$(id -g)
+uidentry=$(getent passwd $myuid)
+
+# If there is no passwd entry for the container UID, attempt to create one
+if [ -z "$uidentry" ] ; then
+    if [ -w /etc/passwd ] ; then
+        echo "$myuid:x:$myuid:$mygid:anonymous uid:$SPARK_HOME:/bin/false" >> /etc/passwd
+    else
+        echo "Container ENTRYPOINT failed to add passwd entry for anonymous UID"
+    fi
+fi
+
+# Execute the container CMD under tini for better hygiene
+/sbin/tini -s -- "$@"
diff --git a/resource-managers/kubernetes/integration-tests-spark-jobs/src/main/scala/org/apache/spark/deploy/kubernetes/integrationtest/jobs/JavaOptionsTest.scala b/resource-managers/kubernetes/integration-tests-spark-jobs/src/main/scala/org/apache/spark/deploy/kubernetes/integrationtest/jobs/JavaOptionsTest.scala
new file mode 100644
index 0000000000000..967032eddccb5
--- /dev/null
+++ b/resource-managers/kubernetes/integration-tests-spark-jobs/src/main/scala/org/apache/spark/deploy/kubernetes/integrationtest/jobs/JavaOptionsTest.scala
@@ -0,0 +1,91 @@
+/*
+ * 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.kubernetes.integrationtest.jobs
+
+import java.io.{File, FileInputStream}
+import java.util.Properties
+
+import com.google.common.collect.Maps
+import scala.collection.JavaConverters._
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.util.Utils
+
+private[spark] object JavaOptionsTest {
+
+  def main(args: Array[String]): Unit = {
+    // scalastyle:off println
+    if (args.length != 2) {
+      println(s"Invalid arguments: ${args.mkString(",")}." +
+        s"Usage: JavaOptionsTest <driver-java-options-list-file> <executor-java-options-list-file>")
+      System.exit(1)
+    }
+    val expectedDriverJavaOptions = loadPropertiesFromFile(args(0))
+    val expectedExecutorJavaOptions = loadPropertiesFromFile(args(1))
+    val nonMatchingDriverOptions = expectedDriverJavaOptions.filter {
+      case (optKey, optValue) => System.getProperty(optKey) != optValue
+    }
+    if (nonMatchingDriverOptions.nonEmpty) {
+      println(s"The driver's JVM options did not match. Expected $expectedDriverJavaOptions." +
+        s" But these options did not match: $nonMatchingDriverOptions.")
+      val sysProps = Maps.fromProperties(System.getProperties).asScala
+      println("Driver system properties are:")
+      for (prop <- sysProps) {
+        println(s"Key: ${prop._1}, Value: ${prop._2}")
+      }
+      System.exit(1)
+    }
+
+    val spark = SparkSession.builder().getOrCreate().sparkContext
+    try {
+      val nonMatchingExecutorOptions = spark.parallelize(Seq(0)).flatMap { _ =>
+        expectedExecutorJavaOptions.filter {
+          case (optKey, optValue) => System.getProperty(optKey) != optValue
+        }
+      }.collectAsMap()
+      if (nonMatchingExecutorOptions.nonEmpty) {
+        val executorSysProps = spark.parallelize(Seq(0)).flatMap { _ =>
+          Maps.fromProperties(System.getProperties).asScala
+        }.collectAsMap()
+        println(s"The executor's JVM options did not match. Expected" +
+          s" $expectedExecutorJavaOptions. But these options did not" +
+          s" match: $nonMatchingExecutorOptions.")
+        println("Executor system properties are:")
+        for (prop <- executorSysProps) {
+          println(s"Key: ${prop._1}, Value: ${prop._2}")
+        }
+      } else {
+        println("All expected JVM options were present on the driver and executors.")
+      }
+    } finally {
+      spark.stop()
+    }
+    // scalastyle:on println
+  }
+
+  private def loadPropertiesFromFile(filePath: String): Map[String, String] = {
+    val file = new File(filePath)
+    if (!file.isFile) {
+      throw new IllegalArgumentException(s"File not found at $filePath or is not a file.")
+    }
+    val properties = new Properties()
+    Utils.tryWithResource(new FileInputStream(file)) { is =>
+      properties.load(is)
+    }
+    Maps.fromProperties(properties).asScala.toMap
+  }
+}
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala
index c6cd6a74c88d1..20e050a25b5cc 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala
@@ -16,9 +16,9 @@
  */
 package org.apache.spark.deploy.kubernetes.integrationtest
 
-import java.io.File
+import java.io.{File, FileOutputStream}
 import java.nio.file.Paths
-import java.util.UUID
+import java.util.{Properties, UUID}
 
 import com.google.common.base.Charsets
 import com.google.common.io.Files
@@ -226,6 +226,36 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
         Seq.empty[String])
   }
 
+  test("Setting JVM options on the driver and executors with spaces.") {
+    assume(testBackend.name == MINIKUBE_TEST_BACKEND)
+    launchStagingServer(SSLOptions(), None)
+    val driverJvmOptionsFile = storeJvmOptionsInTempFile(
+        Map("simpleDriverConf" -> "simpleDriverConfValue",
+            "driverconfwithspaces" -> "driver conf with spaces value"),
+        "driver-jvm-options.properties",
+        "JVM options that should be set on the driver.")
+    val executorJvmOptionsFile = storeJvmOptionsInTempFile(
+        Map("simpleExecutorConf" -> "simpleExecutorConfValue",
+            "executor conf with spaces" -> "executor conf with spaces value"),
+        "executor-jvm-options.properties",
+        "JVM options that should be set on the executors.")
+    sparkConf.set(SparkLauncher.DRIVER_EXTRA_JAVA_OPTIONS,
+        "-DsimpleDriverConf=simpleDriverConfValue" +
+            " -Ddriverconfwithspaces='driver conf with spaces value'")
+    sparkConf.set(SparkLauncher.EXECUTOR_EXTRA_JAVA_OPTIONS,
+        "-DsimpleExecutorConf=simpleExecutorConfValue" +
+            " -D\'executor conf with spaces\'=\'executor conf with spaces value\'")
+    sparkConf.set("spark.files",
+        Seq(driverJvmOptionsFile.getAbsolutePath, executorJvmOptionsFile.getAbsolutePath)
+            .mkString(","))
+    runSparkApplicationAndVerifyCompletion(
+        JavaMainAppResource(SUBMITTER_LOCAL_MAIN_APP_RESOURCE),
+        JAVA_OPTIONS_MAIN_CLASS,
+        Seq(s"All expected JVM options were present on the driver and executors."),
+        Array(driverJvmOptionsFile.getName, executorJvmOptionsFile.getName),
+        Seq.empty[String])
+  }
+
   test("Use a very long application name.") {
     assume(testBackend.name == MINIKUBE_TEST_BACKEND)
 
@@ -339,6 +369,20 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
       }
     }
   }
+
+  private def storeJvmOptionsInTempFile(
+      options: Map[String, String],
+      propertiesFileName: String,
+      comments: String): File = {
+    val tempDir = Utils.createTempDir()
+    val propertiesFile = new File(tempDir, propertiesFileName)
+    val properties = new Properties()
+    options.foreach { case (propKey, propValue) => properties.setProperty(propKey, propValue) }
+    Utils.tryWithResource(new FileOutputStream(propertiesFile)) { os =>
+      properties.store(os, comments)
+    }
+    propertiesFile
+  }
 }
 
 private[spark] object KubernetesSuite {
@@ -368,6 +412,8 @@ private[spark] object KubernetesSuite {
     ".integrationtest.jobs.FileExistenceTest"
   val GROUP_BY_MAIN_CLASS = "org.apache.spark.deploy.kubernetes" +
     ".integrationtest.jobs.GroupByTest"
+  val JAVA_OPTIONS_MAIN_CLASS = "org.apache.spark.deploy.kubernetes" +
+    ".integrationtest.jobs.JavaOptionsTest"
   val TEST_EXISTENCE_FILE_CONTENTS = "contents"
 
   case object ShuffleNotReadyException extends Exception
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/SparkReadinessWatcher.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/SparkReadinessWatcher.scala
index 20517eb2fc2a6..bd604ab94b936 100644
--- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/SparkReadinessWatcher.scala
+++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/SparkReadinessWatcher.scala
@@ -37,5 +37,5 @@ private[spark] class SparkReadinessWatcher[T <: HasMetadata] extends Watcher[T]
 
   override def onClose(cause: KubernetesClientException): Unit = {}
 
-  def waitUntilReady(): Boolean = signal.get(30, TimeUnit.SECONDS)
+  def waitUntilReady(): Boolean = signal.get(60, TimeUnit.SECONDS)
 }