Repository: GoogleCloudPlatform/spark-on-k8s-operator
Pull request #966 (2020-07-06 11:56:19)

volcano scheduler support custom request resource
- support custom resource request for volcano
- add basic e2e case for the feature

Signed-off-by: Thor <1187526662@qq.com>

googlebot: 
Thanks for your pull request. It looks like this may be your first contribution to a Google open source project (if not, look below for help). Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

:memo: **Please visit <https://cla.developers.google.com/> to sign.**

Once you've signed (or fixed any issues), please reply here with `@googlebot I signed it!` and we'll verify it.

----

#### What to do if you already signed the CLA

##### Individual signers

*   It's possible we don't have your GitHub username or you're using a different email address on your commit. Check [your existing CLA data](https://cla.developers.google.com/clas) and verify that your [email is set on your git commits](https://help.github.com/articles/setting-your-email-in-git/).

##### Corporate signers

*   Your company has a Point of Contact who decides which employees are authorized to participate. Ask your POC to be added to the group of authorized contributors. If you don't know who your Point of Contact is, direct the Google project maintainer to [go/cla#troubleshoot](http://go/cla#troubleshoot) ([Public version](https://opensource.google/docs/cla/#troubleshoot)).
*   The email used to register you as an authorized contributor must be the email used for the Git commit. Check [your existing CLA data](https://cla.developers.google.com/clas) and verify that your [email is set on your git commits](https://help.github.com/articles/setting-your-email-in-git/).
*   The email used to register you as an authorized contributor must also be [attached to your GitHub account](https://github.com/settings/emails).
		

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2FGoogleCloudPlatform%2Fspark-on-k8s-operator%2Fpull%2F966) for more info**.

<!-- need_sender_cla -->
Thor-wl: @googlebot I signed it!
googlebot: CLAs look good, thanks!

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2FGoogleCloudPlatform%2Fspark-on-k8s-operator%2Fpull%2F966) for more info**.

<!-- ok -->
k82cn: I'm not sure how this PR handle #964 :)
Thor-wl: > I'm not sure how this PR handle #964 :)

added filed CustomRequestResource in BatchSchedulerConfiguration provides a way to custom config resource reuqest.If configed in the yaml, the total request resource of exectors and drivers will not be calculated

Patch:
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 236602f..b007a3d 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -17,12 +17,12 @@ build:
     - docker login -u $CI_REGISTRY_USER -p $CI_REGISTRY_PASSWORD $CI_REGISTRY ;
     - >
       if [ "${SPARK_REGISTRY}" != "" -a "${SPARK_VERSION}" != "" ] ; then
-        tagStamp=$(git describe --dirty)_${SPARK_VERSION}
+        tagStamp=$(git describe --tags --dirty)_${SPARK_VERSION}
         echo Using SPARK_IMAGE ${SPARK_REGISTRY}:${SPARK_VERSION}
         echo CI_REGISTRY_IMAGE_TAG is ${CI_REGISTRY_IMAGE}/spark-operator:${tagStamp}
         docker build --build-arg SPARK_IMAGE=${SPARK_REGISTRY}:${SPARK_VERSION} -t ${CI_REGISTRY_IMAGE}/spark-operator:${tagStamp} .
       else
-        tagStamp=$(git describe --dirty) ; echo tagStamp is ${tagStamp} ;
+        tagStamp=$(git describe --tags --dirty) ; echo tagStamp is ${tagStamp} ;
         echo CI_REGISTRY_IMAGE_TAG is ${CI_REGISTRY_IMAGE}/spark-operator:${tagStamp}
         docker build -t ${CI_REGISTRY_IMAGE}/spark-operator:${tagStamp} .
       fi
diff --git a/Dockerfile.rh b/Dockerfile.rh
index 75e7325..42880b3 100644
--- a/Dockerfile.rh
+++ b/Dockerfile.rh
@@ -20,7 +20,7 @@
 # 1. Your Docker version is >= 18.09.3
 # 2. export DOCKER_BUILDKIT=1
 
-ARG SPARK_IMAGE=gcr.io/spark-operator/spark:v2.4.5
+ARG SPARK_IMAGE=gcr.io/spark-operator/spark:v3.0.0
 
 FROM golang:1.14.0-alpine as builder
 
diff --git a/README.md b/README.md
index e0eab8c..5ead7ec 100644
--- a/README.md
+++ b/README.md
@@ -16,7 +16,7 @@
 
 **If you are currently using the `v1beta1` version of the APIs in your manifests, please update them to use the `v1beta2` version by changing `apiVersion: "sparkoperator.k8s.io/<version>"` to `apiVersion: "sparkoperator.k8s.io/v1beta2"`. You will also need to delete the `previous` version of the CustomResourceDefinitions named `sparkapplications.sparkoperator.k8s.io` and `scheduledsparkapplications.sparkoperator.k8s.io`, and replace them with the `v1beta2` version either by installing the latest version of the operator or by running `kubectl create -f manifest/crds`.**
 
-Customization of Spark pods, e.g., mounting arbitrary volumes and setting pod affinity, is implemented using a Kubernetes [Mutating Admission Webhook](https://kubernetes.io/docs/reference/access-authn-authz/extensible-admission-controllers/), which became beta in Kubernetes 1.9. The mutating admission webhook is enabled by default if you install the operator using the Helm [chart](https://github.com/helm/charts/tree/master/incubator/sparkoperator). Check out the [Quick Start Guide](docs/quick-start-guide.md#using-the-mutating-admission-webhook)  on how to enable the webhook.
+Customization of Spark pods, e.g., mounting arbitrary volumes and setting pod affinity, is implemented using a Kubernetes [Mutating Admission Webhook](https://kubernetes.io/docs/reference/access-authn-authz/extensible-admission-controllers/), which became beta in Kubernetes 1.9. The mutating admission webhook is disabled by default if you install the operator using the Helm [chart](https://github.com/helm/charts/tree/master/incubator/sparkoperator). Check out the [Quick Start Guide](docs/quick-start-guide.md#using-the-mutating-admission-webhook) on how to enable the webhook.
 
 ## Prerequisites
 
@@ -45,7 +45,8 @@ The following table lists the most recent few versions of the operator.
 
 | Operator Version | API Version | Kubernetes Version | Base Spark Version | Operator Image Tag |
 | ------------- | ------------- | ------------- | ------------- | ------------- |
-| `latest` (master HEAD) | `v1beta2` | 1.13+ | `2.4.5` | `latest` |
+| `latest` (master HEAD) | `v1beta2` | 1.13+ | `3.0.0` | `latest` |
+| `v1beta2-1.2.0-3.0.0` | `v1beta2` | 1.13+ | `3.0.0` | `v1beta2-1.2.0-3.0.0` |
 | `v1beta2-1.1.2-2.4.5` | `v1beta2` | 1.13+ | `2.4.5` | `v1beta2-1.1.2-2.4.5` |
 | `v1beta2-1.0.1-2.4.4` | `v1beta2` | 1.13+ | `2.4.4` | `v1beta2-1.0.1-2.4.4` | 
 | `v1beta2-1.0.0-2.4.4` | `v1beta2` | 1.13+ | `2.4.4` | `v1beta2-1.0.0-2.4.4` |
diff --git a/docs/api-docs.md b/docs/api-docs.md
index 0cd97d0..a0ff575 100644
--- a/docs/api-docs.md
+++ b/docs/api-docs.md
@@ -586,6 +586,21 @@ SparkUIConfiguration
 <p>SparkUIOptions allows configuring the Service and the Ingress to expose the sparkUI</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>dynamicAllocation</code></br>
+<em>
+<a href="#sparkoperator.k8s.io/v1beta2.DynamicAllocation">
+DynamicAllocation
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>DynamicAllocation configures dynamic allocation that becomes available for the Kubernetes
+scheduleer backend since Spark 3.0.</p>
+</td>
+</tr>
 </table>
 </td>
 </tr>
@@ -969,6 +984,86 @@ executors to connect to the driver.</p>
 <p>
 <p>DriverState tells the current state of a spark driver.</p>
 </p>
+<h3 id="sparkoperator.k8s.io/v1beta2.DynamicAllocation">DynamicAllocation
+</h3>
+<p>
+(<em>Appears on:</em>
+<a href="#sparkoperator.k8s.io/v1beta2.SparkApplicationSpec">SparkApplicationSpec</a>)
+</p>
+<p>
+<p>DynamicAllocation contains configuration options for dynamic allocation.</p>
+</p>
+<table>
+<thead>
+<tr>
+<th>Field</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>enabled</code></br>
+<em>
+bool
+</em>
+</td>
+<td>
+<p>Enabled controls whether dynamic allocation is enabled or not.</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>initialExecutors</code></br>
+<em>
+int32
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>InitialExecutors is the initial number of executors to request. If .spec.executor.instances
+is also set, the initial number of executors is set to the bigger of that and this option.</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>minExecutors</code></br>
+<em>
+int32
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>MinExecutors is the lower bound for the number of executors if dynamic allocation is enabled.</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>maxExecutors</code></br>
+<em>
+int32
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>MaxExecutors is the upper bound for the number of executors if dynamic allocation is enabled.</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>shuffleTrackingTimeout</code></br>
+<em>
+int64
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>ShuffleTrackingTimeout controls the timeout in milliseconds for executors that are holding
+shuffle data if shuffle tracking is enabled (true by default if dynamic allocation is enabled).</p>
+</td>
+</tr>
+</tbody>
+</table>
 <h3 id="sparkoperator.k8s.io/v1beta2.ExecutorSpec">ExecutorSpec
 </h3>
 <p>
@@ -2072,6 +2167,21 @@ SparkUIConfiguration
 <p>SparkUIOptions allows configuring the Service and the Ingress to expose the sparkUI</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>dynamicAllocation</code></br>
+<em>
+<a href="#sparkoperator.k8s.io/v1beta2.DynamicAllocation">
+DynamicAllocation
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>DynamicAllocation configures dynamic allocation that becomes available for the Kubernetes
+scheduleer backend since Spark 3.0.</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="sparkoperator.k8s.io/v1beta2.SparkApplicationStatus">SparkApplicationStatus
@@ -2582,7 +2692,7 @@ string
 <a href="#sparkoperator.k8s.io/v1beta2.SparkApplicationSpec">SparkApplicationSpec</a>)
 </p>
 <p>
-<p>Specific SparkUI config parameters</p>
+<p>SparkUIConfiguration is for driver UI specific configuration parameters.</p>
 </p>
 <table>
 <thead>
@@ -2636,5 +2746,5 @@ map[string]string
 <hr/>
 <p><em>
 Generated with <code>gen-crd-api-reference-docs</code>
-on git commit <code>f313873</code>.
+on git commit <code>555c27a</code>.
 </em></p>
diff --git a/docs/developer-guide.md b/docs/developer-guide.md
index e7b4323..8878714 100644
--- a/docs/developer-guide.md
+++ b/docs/developer-guide.md
@@ -10,7 +10,7 @@ The easiest way to build the operator without worrying about its dependencies is
 $ docker build -t <image-tag> .
 ```
 
-The operator image is built upon a base Spark image that defaults to `gcr.io/spark-operator/spark:v2.4.5`. If you want to use your own Spark image (e.g., an image with a different version of Spark or some custom dependencies), specify the argument `SPARK_IMAGE` as the following example shows: 
+The operator image is built upon a base Spark image that defaults to `gcr.io/spark-operator/spark:v3.0.0`. If you want to use your own Spark image (e.g., an image with a different version of Spark or some custom dependencies), specify the argument `SPARK_IMAGE` as the following example shows: 
 
 ```bash
 $ docker build --build-arg SPARK_IMAGE=<your Spark image> -t <image-tag> .
diff --git a/docs/quick-start-guide.md b/docs/quick-start-guide.md
index 814bd25..eb1f47e 100644
--- a/docs/quick-start-guide.md
+++ b/docs/quick-start-guide.md
@@ -88,8 +88,8 @@ spec:
     labels:
       version: 2.3.0
     memory: 512m
-  image: gcr.io/ynli-k8s/spark:v2.4.5
-  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.11-2.3.0.jar
+  image: gcr.io/ynli-k8s/spark:v3.0.0
+  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-2.3.0.jar
   mainClass: org.apache.spark.examples.SparkPi
   mode: cluster
   restartPolicy:
diff --git a/docs/user-guide.md b/docs/user-guide.md
index ed39a52..a0d2be6 100644
--- a/docs/user-guide.md
+++ b/docs/user-guide.md
@@ -34,6 +34,7 @@ The Kubernetes Operator for Apache Spark ships with a command-line tool called `
     * [Using Container LifeCycle Hooks](#using-container-lifecycle-hooks)
     * [Python Support](#python-support)
     * [Monitoring](#monitoring)
+    * [Dynamic Allocation](#dynamic-allocation)
 * [Working with SparkApplications](#working-with-sparkapplications)
     * [Creating a New SparkApplication](#creating-a-new-sparkapplication)
     * [Deleting a SparkApplication](#deleting-a-sparkapplication)
@@ -69,9 +70,9 @@ metadata:
 spec:
   type: Scala
   mode: cluster
-  image: gcr.io/spark/spark:v2.4.5
+  image: gcr.io/spark/spark:v3.0.0
   mainClass: org.apache.spark.examples.SparkPi
-  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.11-2.4.5.jar
+  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.0.0.jar
 ```
 
 ### Specifying Application Dependencies
@@ -135,7 +136,7 @@ spec:
     coreLimit: 200m
     memory: 512m
     labels:
-      version: 2.4.5
+      version: 3.0.0
     serviceAccount: spark
 ```
 
@@ -155,7 +156,7 @@ spec:
     instances: 1
     memory: 512m
     labels:
-      version: 2.4.5
+      version: 3.0.0
 ```
 
 ### Specifying Extra Java Options
@@ -236,7 +237,7 @@ spec:
       name: "amd.com/gpu"   # GPU resource name
       quantity: 1           # number of GPUs to request
     labels:
-      version: 2.4.5
+      version: 3.0.0
     serviceAccount: spark
   executor:
     cores: 1
@@ -260,7 +261,7 @@ spec:
     memory: "512m"
     hostNetwork: true
     labels:
-      version: 2.4.5
+      version: 3.0.0
     serviceAccount: spark
   executor:
     cores: 1
@@ -621,7 +622,7 @@ Note that Python binding for PySpark is available in Apache Spark 2.4.
 
 The operator supports using the Spark metric system to expose metrics to a variety of sinks. Particularly, it is able to automatically configure the metric system to expose metrics to [Prometheus](https://prometheus.io/). Specifically, the field `.spec.monitoring` specifies how application monitoring is handled and particularly how metrics are to be reported. The metric system is configured through the configuration file `metrics.properties`, which gets its content from the field `.spec.monitoring.metricsProperties`. The content of [metrics.properties](../spark-docker/conf/metrics.properties) will be used by default if `.spec.monitoring.metricsProperties` is not specified. `.spec.monitoring.metricsPropertiesFile` overwrite the value `spark.metrics.conf` in spark.properties, and will not use content from `.spec.monitoring.metricsProperties`. You can choose to enable or disable reporting driver and executor metrics using the fields `.spec.monitoring.exposeDriverMetrics` and `.spec.monitoring.exposeExecutorMetrics`, respectively.
 
-Further, the field `.spec.monitoring.prometheus` specifies how metrics are exposed to Prometheus using the [Prometheus JMX exporter](https://github.com/prometheus/jmx_exporter). When `.spec.monitoring.prometheus` is specified, the operator automatically configures the JMX exporter to run as a Java agent. The only required field of `.spec.monitoring.prometheus` is `jmxExporterJar`, which specified the path to the Prometheus JMX exporter Java agent jar in the container. If you use the image `gcr.io/spark-operator/spark:v2.4.5-gcs-prometheus`, the jar is located at `/prometheus/jmx_prometheus_javaagent-0.11.0.jar`. The field `.spec.monitoring.prometheus.port` specifies the port the JMX exporter Java agent binds to and defaults to `8090` if not specified. The field `.spec.monitoring.prometheus.configuration` specifies the content of the configuration to be used with the JMX exporter. The content of [prometheus.yaml](../spark-docker/conf/prometheus.yaml) will be used by default if `.spec.monitoring.prometheus.configuration` is not specified.    
+Further, the field `.spec.monitoring.prometheus` specifies how metrics are exposed to Prometheus using the [Prometheus JMX exporter](https://github.com/prometheus/jmx_exporter). When `.spec.monitoring.prometheus` is specified, the operator automatically configures the JMX exporter to run as a Java agent. The only required field of `.spec.monitoring.prometheus` is `jmxExporterJar`, which specified the path to the Prometheus JMX exporter Java agent jar in the container. If you use the image `gcr.io/spark-operator/spark:v3.0.0-gcs-prometheus`, the jar is located at `/prometheus/jmx_prometheus_javaagent-0.11.0.jar`. The field `.spec.monitoring.prometheus.port` specifies the port the JMX exporter Java agent binds to and defaults to `8090` if not specified. The field `.spec.monitoring.prometheus.configuration` specifies the content of the configuration to be used with the JMX exporter. The content of [prometheus.yaml](../spark-docker/conf/prometheus.yaml) will be used by default if `.spec.monitoring.prometheus.configuration` is not specified.    
 
 Below is an example that shows how to configure the metric system to expose metrics to Prometheus using the Prometheus JMX exporter. Note that the JMX exporter Java agent jar is listed as a dependency and will be downloaded to where `.spec.dep.jarsDownloadDir` points to in Spark 2.3.x, which is `/var/spark-data/spark-jars` by default. Things are different in Spark 2.4 as dependencies will be downloaded to the local working directory instead in Spark 2.4. A complete example can be found in [examples/spark-pi-prometheus.yaml](../examples/spark-pi-prometheus.yaml).
 
@@ -638,6 +639,21 @@ spec:
 
 The operator automatically adds the annotations such as `prometheus.io/scrape=true` on the driver and/or executor pods (depending on the values of  `.spec.monitoring.exposeDriverMetrics` and `.spec.monitoring.exposeExecutorMetrics`) so the metrics exposed on the pods can be scraped by the Prometheus server in the same cluster.
 
+### Dynamic Allocation
+
+The operator supports a limited form of [Spark Dynamic Resource Allocation](http://spark.apache.org/docs/latest/job-scheduling.html#dynamic-resource-allocation) through the shuffle tracking enhancement introduced in Spark 3.0.0 *without needing an external shuffle service* (not available in the Kubernetes mode). See this [issue](https://issues.apache.org/jira/browse/SPARK-27963) for detais on the enhancement. To enable this limited form of dynamic allocation, follow the example below:
+
+```yaml
+spec:
+  dynamicAllocation:
+    enabled: true
+    initialExecutors: 2
+    minExecutors: 2
+    maxExecutors: 10
+```
+
+Note that if dynamic allocation is enabled, the number of executors to request initially is set to the bigger of `.spec.dynamicAllocation.initialExecutors` and `.spec.executor.instances` if both are set.
+
 ## Working with SparkApplications
 
 ### Creating a New SparkApplication
@@ -709,9 +725,9 @@ spec:
   template:
     type: Scala
     mode: cluster
-    image: gcr.io/spark/spark:v2.4.5
+    image: gcr.io/spark/spark:v3.0.0
     mainClass: org.apache.spark.examples.SparkPi
-    mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.11-2.3.0.jar
+    mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-2.3.0.jar
     driver:
       cores: 1
       memory: 512m
diff --git a/docs/volcano-integration.md b/docs/volcano-integration.md
index 061f5f5..eb0f6c3 100644
--- a/docs/volcano-integration.md
+++ b/docs/volcano-integration.md
@@ -31,11 +31,11 @@ metadata:
 spec:
   type: Scala
   mode: cluster
-  image: "gcr.io/spark-operator/spark:v2.4.5"
+  image: "gcr.io/spark-operator/spark:v3.0.0"
   imagePullPolicy: Always
   mainClass: org.apache.spark.examples.SparkPi
-  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.11-2.4.5.jar"
-  sparkVersion: "2.4.5"
+  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.12-3.0.0.jar"
+  sparkVersion: "3.0.0"
   batchScheduler: "volcano"   #Note: the batch scheduler name must be specified with `volcano`
   restartPolicy:
     type: Never
@@ -49,7 +49,7 @@ spec:
     coreLimit: "1200m"
     memory: "512m"        
     labels:
-      version: 2.4.5
+      version: 3.0.0
     serviceAccount: spark
     volumeMounts:
       - name: "test-volume"
@@ -59,7 +59,7 @@ spec:
     instances: 1
     memory: "512m"    
     labels:
-      version: 2.4.5
+      version: 3.0.0
     volumeMounts:
       - name: "test-volume"
         mountPath: "/tmp"
@@ -78,7 +78,7 @@ If SparkApplication is configured to run with Volcano, there are some details un
 1. Kubernetes Operator for Apache Spark's webhook will patch pods' `schedulerName` according to the `batchScheduler` in SparkApplication Spec.
 2. Before submitting spark application, Kubernetes Operator for Apache Spark will create a Volcano native resource 
    `PodGroup`[here](https://github.com/volcano-sh/volcano/blob/a8fb05ce6c6902e366cb419d6630d66fc759121e/pkg/apis/scheduling/v1alpha2/types.go#L93) for the whole application.
-   and as a brief introduction，most of the Volcano's advanced scheduling features, such as pod delay creation, resource fairness and gang scheduling are all depend on this resource. 
+   and as a brief introduction, most of the Volcano's advanced scheduling features, such as pod delay creation, resource fairness and gang scheduling are all depend on this resource. 
    Also a new pod annotation named `scheduling.k8s.io/group-name` will be added.
 3. Volcano scheduler will take over all of the pods that both have schedulerName and annotation correctly configured for scheduling.
 
diff --git a/examples/spark-pi-configmap.yaml b/examples/spark-pi-configmap.yaml
index 56016ed..07200fc 100644
--- a/examples/spark-pi-configmap.yaml
+++ b/examples/spark-pi-configmap.yaml
@@ -21,11 +21,11 @@ metadata:
 spec:
   type: Scala
   mode: cluster
-  image: "gcr.io/spark-operator/spark:v2.4.5"
+  image: "gcr.io/spark-operator/spark:v3.0.0"
   imagePullPolicy: Always
   mainClass: org.apache.spark.examples.SparkPi
-  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.11-2.4.5.jar"
-  sparkVersion: "2.4.5"
+  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.12-3.0.0.jar"
+  sparkVersion: "3.0.0"
   restartPolicy:
     type: Never
   volumes:
@@ -37,7 +37,7 @@ spec:
     coreLimit: "1200m"
     memory: "512m"
     labels:
-      version: 2.4.5
+      version: 3.0.0
     serviceAccount: spark
     volumeMounts:
       - name: config-vol
@@ -47,7 +47,7 @@ spec:
     instances: 1
     memory: "512m"
     labels:
-      version: 2.4.5
+      version: 3.0.0
     volumeMounts:
       - name: config-vol
         mountPath: /opt/spark/mycm
diff --git a/examples/spark-pi-custom-resource.yaml b/examples/spark-pi-custom-resource.yaml
new file mode 100644
index 0000000..3f27bd7
--- /dev/null
+++ b/examples/spark-pi-custom-resource.yaml
@@ -0,0 +1,58 @@
+#
+# Copyright 2017 Google LLC
+#
+# Licensed 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
+#
+#     https://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.
+
+apiVersion: "sparkoperator.k8s.io/v1beta2"
+kind: SparkApplication
+metadata:
+  name: spark-pi-custom-resource
+  namespace: default
+spec:
+  type: Scala
+  mode: cluster
+  image: "gcr.io/spark-operator/spark:v2.4.5"
+  imagePullPolicy: Always
+  mainClass: org.apache.spark.examples.SparkPi
+  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.11-2.4.5.jar"
+  sparkVersion: "2.4.5"
+  restartPolicy:
+    type: Never
+  volumes:
+    - name: "test-volume"
+      hostPath:
+        path: "/tmp"
+        type: Directory
+  driver:
+    cores: 1
+    coreLimit: "1200m"
+    memory: "512m"
+    labels:
+      version: 2.4.5
+    serviceAccount: spark
+    volumeMounts:
+      - name: "test-volume"
+        mountPath: "/tmp"
+  executor:
+    cores: 1
+    instances: 1
+    memory: "512m"
+    labels:
+      version: 2.4.5
+    volumeMounts:
+      - name: "test-volume"
+        mountPath: "/tmp"
+  batchSchedulerOptions:
+    resources:
+        cpu: "2"
+        memory: "4096m"
\ No newline at end of file
diff --git a/examples/spark-pi-prometheus.yaml b/examples/spark-pi-prometheus.yaml
index 68d75d2..bd936aa 100644
--- a/examples/spark-pi-prometheus.yaml
+++ b/examples/spark-pi-prometheus.yaml
@@ -22,13 +22,13 @@ metadata:
 spec:
   type: Scala
   mode: cluster
-  image: "gcr.io/spark-operator/spark:v2.4.5-gcs-prometheus"
+  image: "gcr.io/spark-operator/spark:v3.0.0-gcs-prometheus"
   imagePullPolicy: Always
   mainClass: org.apache.spark.examples.SparkPi
-  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.11-2.4.5.jar"
+  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.12-3.0.0.jar"
   arguments:
     - "100000"
-  sparkVersion: "2.4.5"
+  sparkVersion: "3.0.0"
   restartPolicy:
     type: Never
   driver:
@@ -36,14 +36,14 @@ spec:
     coreLimit: "1200m"
     memory: "512m"
     labels:
-      version: 2.4.5
+      version: 3.0.0
     serviceAccount: spark
   executor:
     cores: 1
     instances: 1
     memory: "512m"
     labels:
-      version: 2.4.5
+      version: 3.0.0
   monitoring:
     exposeDriverMetrics: true
     exposeExecutorMetrics: true
diff --git a/examples/spark-pi-schedule.yaml b/examples/spark-pi-schedule.yaml
index a95d3e1..f508c59 100644
--- a/examples/spark-pi-schedule.yaml
+++ b/examples/spark-pi-schedule.yaml
@@ -25,11 +25,11 @@ spec:
   template:
     type: Scala
     mode: cluster
-    image: "gcr.io/spark-operator/spark:v2.4.5"
+    image: "gcr.io/spark-operator/spark:v3.0.0"
     imagePullPolicy: Always
     mainClass: org.apache.spark.examples.SparkPi
-    mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.11-2.4.5.jar"
-    sparkVersion: "2.4.5"
+    mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.12-3.0.0.jar"
+    sparkVersion: "3.0.0"
     restartPolicy:
       type: Never
     driver:
@@ -37,11 +37,11 @@ spec:
       coreLimit: "1200m"
       memory: "512m"
       labels:
-        version: 2.4.5
+        version: 3.0.0
       serviceAccount: spark
     executor:
       cores: 1
       instances: 1
       memory: "512m"
       labels:
-        version: 2.4.5 
+        version: 3.0.0 
diff --git a/examples/spark-pi.yaml b/examples/spark-pi.yaml
index 5b62597..e0ad2d1 100644
--- a/examples/spark-pi.yaml
+++ b/examples/spark-pi.yaml
@@ -21,11 +21,11 @@ metadata:
 spec:
   type: Scala
   mode: cluster
-  image: "gcr.io/spark-operator/spark:v2.4.5"
+  image: "gcr.io/spark-operator/spark:v3.0.0"
   imagePullPolicy: Always
   mainClass: org.apache.spark.examples.SparkPi
-  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.11-2.4.5.jar"
-  sparkVersion: "2.4.5"
+  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.12-3.0.0.jar"
+  sparkVersion: "3.0.0"
   restartPolicy:
     type: Never
   volumes:
@@ -38,7 +38,7 @@ spec:
     coreLimit: "1200m"
     memory: "512m"
     labels:
-      version: 2.4.5
+      version: 3.0.0
     serviceAccount: spark
     volumeMounts:
       - name: "test-volume"
@@ -48,7 +48,7 @@ spec:
     instances: 1
     memory: "512m"
     labels:
-      version: 2.4.5
+      version: 3.0.0
     volumeMounts:
       - name: "test-volume"
         mountPath: "/tmp"
diff --git a/examples/spark-py-pi.yaml b/examples/spark-py-pi.yaml
index 5f644d4..ee8639a 100644
--- a/examples/spark-py-pi.yaml
+++ b/examples/spark-py-pi.yaml
@@ -25,10 +25,10 @@ spec:
   type: Python
   pythonVersion: "2"
   mode: cluster
-  image: "gcr.io/spark-operator/spark-py:v2.4.5"
+  image: "gcr.io/spark-operator/spark-py:v3.0.0"
   imagePullPolicy: Always
   mainApplicationFile: local:///opt/spark/examples/src/main/python/pi.py
-  sparkVersion: "2.4.5"
+  sparkVersion: "3.0.0"
   restartPolicy:
     type: OnFailure
     onFailureRetries: 3
@@ -40,11 +40,11 @@ spec:
     coreLimit: "1200m"
     memory: "512m"
     labels:
-      version: 2.4.5
+      version: 3.0.0
     serviceAccount: spark
   executor:
     cores: 1
     instances: 1
     memory: "512m"
     labels:
-      version: 2.4.5
+      version: 3.0.0
diff --git a/manifest/crds/sparkoperator.k8s.io_scheduledsparkapplications.yaml b/manifest/crds/sparkoperator.k8s.io_scheduledsparkapplications.yaml
index 2a9ce43..4df3582 100644
--- a/manifest/crds/sparkoperator.k8s.io_scheduledsparkapplications.yaml
+++ b/manifest/crds/sparkoperator.k8s.io_scheduledsparkapplications.yaml
@@ -775,6 +775,7 @@ spec:
                                   type: string
                               required:
                               - containerPort
+                              - protocol
                               type: object
                             type: array
                             x-kubernetes-list-map-keys:
@@ -1507,6 +1508,7 @@ spec:
                                   type: string
                               required:
                               - containerPort
+                              - protocol
                               type: object
                             type: array
                             x-kubernetes-list-map-keys:
@@ -1800,6 +1802,23 @@ spec:
                         type: object
                       type: array
                   type: object
+                dynamicAllocation:
+                  properties:
+                    enabled:
+                      type: boolean
+                    initialExecutors:
+                      format: int32
+                      type: integer
+                    maxExecutors:
+                      format: int32
+                      type: integer
+                    minExecutors:
+                      format: int32
+                      type: integer
+                    shuffleTrackingTimeout:
+                      format: int64
+                      type: integer
+                  type: object
                 executor:
                   properties:
                     affinity:
@@ -2505,6 +2524,7 @@ spec:
                                   type: string
                               required:
                               - containerPort
+                              - protocol
                               type: object
                             type: array
                             x-kubernetes-list-map-keys:
@@ -3129,6 +3149,7 @@ spec:
                                   type: string
                               required:
                               - containerPort
+                              - protocol
                               type: object
                             type: array
                             x-kubernetes-list-map-keys:
diff --git a/manifest/crds/sparkoperator.k8s.io_sparkapplications.yaml b/manifest/crds/sparkoperator.k8s.io_sparkapplications.yaml
index 9feda0e..9223270 100644
--- a/manifest/crds/sparkoperator.k8s.io_sparkapplications.yaml
+++ b/manifest/crds/sparkoperator.k8s.io_sparkapplications.yaml
@@ -761,6 +761,7 @@ spec:
                               type: string
                           required:
                           - containerPort
+                          - protocol
                           type: object
                         type: array
                         x-kubernetes-list-map-keys:
@@ -1493,6 +1494,7 @@ spec:
                               type: string
                           required:
                           - containerPort
+                          - protocol
                           type: object
                         type: array
                         x-kubernetes-list-map-keys:
@@ -1786,6 +1788,23 @@ spec:
                     type: object
                   type: array
               type: object
+            dynamicAllocation:
+              properties:
+                enabled:
+                  type: boolean
+                initialExecutors:
+                  format: int32
+                  type: integer
+                maxExecutors:
+                  format: int32
+                  type: integer
+                minExecutors:
+                  format: int32
+                  type: integer
+                shuffleTrackingTimeout:
+                  format: int64
+                  type: integer
+              type: object
             executor:
               properties:
                 affinity:
@@ -2491,6 +2510,7 @@ spec:
                               type: string
                           required:
                           - containerPort
+                          - protocol
                           type: object
                         type: array
                         x-kubernetes-list-map-keys:
@@ -3115,6 +3135,7 @@ spec:
                               type: string
                           required:
                           - containerPort
+                          - protocol
                           type: object
                         type: array
                         x-kubernetes-list-map-keys:
diff --git a/manifest/spark-operator-with-metrics.yaml b/manifest/spark-operator-with-metrics.yaml
index b668543..8130805 100644
--- a/manifest/spark-operator-with-metrics.yaml
+++ b/manifest/spark-operator-with-metrics.yaml
@@ -21,13 +21,13 @@ metadata:
   namespace: spark-operator
   labels:
     app.kubernetes.io/name: sparkoperator
-    app.kubernetes.io/version: v2.4.5-v1beta2
+    app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
 spec:
   replicas: 1
   selector:
     matchLabels:
       app.kubernetes.io/name: sparkoperator
-      app.kubernetes.io/version: v2.4.5-v1beta2
+      app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
   strategy:
     type: Recreate
   template:
@@ -38,12 +38,12 @@ spec:
         prometheus.io/path: "/metrics"
       labels:
         app.kubernetes.io/name: sparkoperator
-        app.kubernetes.io/version: v2.4.5-v1beta2
+        app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
     spec:
       serviceAccountName: sparkoperator
       containers:
       - name: sparkoperator
-        image: gcr.io/spark-operator/spark-operator:v2.4.5-v1beta2-latest
+        image: gcr.io/spark-operator/spark-operator:v1beta2-1.2.0-3.0.0
         imagePullPolicy: Always
         ports:
           - containerPort: 10254
diff --git a/manifest/spark-operator-with-webhook.yaml b/manifest/spark-operator-with-webhook.yaml
index b2f292f..c206243 100644
--- a/manifest/spark-operator-with-webhook.yaml
+++ b/manifest/spark-operator-with-webhook.yaml
@@ -21,20 +21,20 @@ metadata:
   namespace: spark-operator
   labels:
     app.kubernetes.io/name: sparkoperator
-    app.kubernetes.io/version: v2.4.5-v1beta2
+    app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
 spec:
   replicas: 1
   selector:
     matchLabels:
       app.kubernetes.io/name: sparkoperator
-      app.kubernetes.io/version: v2.4.5-v1beta2
+      app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
   strategy:
     type: Recreate
   template:
     metadata:
       labels:
         app.kubernetes.io/name: sparkoperator
-        app.kubernetes.io/version: v2.4.5-v1beta2
+        app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
     spec:
       serviceAccountName: sparkoperator
       volumes:
@@ -43,7 +43,7 @@ spec:
           secretName: spark-webhook-certs
       containers:
       - name: sparkoperator
-        image: gcr.io/spark-operator/spark-operator:v2.4.5-v1beta2-latest
+        image: gcr.io/spark-operator/spark-operator:v1beta2-1.2.0-3.0.0
         imagePullPolicy: Always
         volumeMounts:
         - name: webhook-certs
@@ -62,20 +62,20 @@ metadata:
   namespace: spark-operator
   labels:
     app.kubernetes.io/name: sparkoperator
-    app.kubernetes.io/version: v2.4.5-v1beta2
+    app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
 spec:
   backoffLimit: 3
   template:
     metadata:
       labels:
         app.kubernetes.io/name: sparkoperator
-        app.kubernetes.io/version: v2.4.5-v1beta2
+        app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
     spec:
       serviceAccountName: sparkoperator
       restartPolicy: Never
       containers:
       - name: main
-        image: gcr.io/spark-operator/spark-operator:v2.4.5-v1beta2-latest
+        image: gcr.io/spark-operator/spark-operator:v1beta2-1.2.0-3.0.0
         imagePullPolicy: IfNotPresent
         command: ["/usr/bin/gencerts.sh", "-p"]
 ---
@@ -91,4 +91,4 @@ spec:
       name: webhook
   selector:
     app.kubernetes.io/name: sparkoperator
-    app.kubernetes.io/version: v2.4.5-v1beta2
+    app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
diff --git a/manifest/spark-operator.yaml b/manifest/spark-operator.yaml
index 589b1b5..8b827a2 100644
--- a/manifest/spark-operator.yaml
+++ b/manifest/spark-operator.yaml
@@ -21,25 +21,25 @@ metadata:
   namespace: spark-operator
   labels:
     app.kubernetes.io/name: sparkoperator
-    app.kubernetes.io/version: v2.4.5-v1beta2
+    app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
 spec:
   replicas: 1
   selector:
     matchLabels:
       app.kubernetes.io/name: sparkoperator
-      app.kubernetes.io/version: v2.4.5-v1beta2
+      app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
   strategy:
     type: Recreate
   template:
     metadata:
       labels:
         app.kubernetes.io/name: sparkoperator
-        app.kubernetes.io/version: v2.4.5-v1beta2
+        app.kubernetes.io/version: v1beta2-1.2.0-3.0.0
     spec:
       serviceAccountName: sparkoperator
       containers:
       - name: sparkoperator
-        image: gcr.io/spark-operator/spark-operator:v2.4.5-v1beta2-latest
+        image: gcr.io/spark-operator/spark-operator:v1beta2-1.2.0-3.0.0
         imagePullPolicy: Always
         args:
         - -logtostderr
diff --git a/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go b/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
index b84db8b..c14c7ba 100644
--- a/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
+++ b/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
@@ -276,6 +276,10 @@ type SparkApplicationSpec struct {
 	// SparkUIOptions allows configuring the Service and the Ingress to expose the sparkUI
 	// +optional
 	SparkUIOptions *SparkUIConfiguration `json:"sparkUIOptions,omitempty"`
+	// DynamicAllocation configures dynamic allocation that becomes available for the Kubernetes
+	// scheduleer backend since Spark 3.0.
+	// +optional
+	DynamicAllocation *DynamicAllocation `json:"dynamicAllocation,omitempty"`
 }
 
 // BatchSchedulerConfiguration used to configure how to batch scheduling Spark Application
@@ -286,9 +290,13 @@ type BatchSchedulerConfiguration struct {
 	// PriorityClassName stands for the name of k8s PriorityClass resource, it's being used in Volcano batch scheduler.
 	// +optional
 	PriorityClassName *string `json:"priorityClassName,omitempty"`
+	// Resources stands for the resource list custom request for. Usually it is used to define the lower-bound limit.
+	// If specified, volcano scheduler will consider it as the resources requested.
+	// +optional
+	Resources apiv1.ResourceList `json:"resources,omitempty"`
 }
 
-// Specific SparkUI config parameters
+// SparkUIConfiguration is for driver UI specific configuration parameters.
 type SparkUIConfiguration struct {
 	// ServicePort allows configuring the port at service level that might be different from the targetPort.
 	// TargetPort should be the same as the one defined in spark.ui.port
@@ -631,6 +639,26 @@ type GPUSpec struct {
 	Quantity int64 `json:"quantity"`
 }
 
+// DynamicAllocation contains configuration options for dynamic allocation.
+type DynamicAllocation struct {
+	// Enabled controls whether dynamic allocation is enabled or not.
+	Enabled bool `json:"enabled,omitempty"`
+	// InitialExecutors is the initial number of executors to request. If .spec.executor.instances
+	// is also set, the initial number of executors is set to the bigger of that and this option.
+	// +optional
+	InitialExecutors *int32 `json:"initialExecutors,omitempty"`
+	// MinExecutors is the lower bound for the number of executors if dynamic allocation is enabled.
+	// +optional
+	MinExecutors *int32 `json:"minExecutors,omitempty"`
+	// MaxExecutors is the upper bound for the number of executors if dynamic allocation is enabled.
+	// +optional
+	MaxExecutors *int32 `json:"maxExecutors,omitempty"`
+	// ShuffleTrackingTimeout controls the timeout in milliseconds for executors that are holding
+	// shuffle data if shuffle tracking is enabled (true by default if dynamic allocation is enabled).
+	// +optional
+	ShuffleTrackingTimeout *int64 `json:"shuffleTrackingTimeout,omitempty"`
+}
+
 // PrometheusMonitoringEnabled returns if Prometheus monitoring is enabled or not.
 func (s *SparkApplication) PrometheusMonitoringEnabled() bool {
 	return s.Spec.Monitoring != nil && s.Spec.Monitoring.Prometheus != nil
diff --git a/pkg/apis/sparkoperator.k8s.io/v1beta2/zz_generated.deepcopy.go b/pkg/apis/sparkoperator.k8s.io/v1beta2/zz_generated.deepcopy.go
index ebd732c..3b63947 100644
--- a/pkg/apis/sparkoperator.k8s.io/v1beta2/zz_generated.deepcopy.go
+++ b/pkg/apis/sparkoperator.k8s.io/v1beta2/zz_generated.deepcopy.go
@@ -166,6 +166,42 @@ func (in *DriverSpec) DeepCopy() *DriverSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *DynamicAllocation) DeepCopyInto(out *DynamicAllocation) {
+	*out = *in
+	if in.InitialExecutors != nil {
+		in, out := &in.InitialExecutors, &out.InitialExecutors
+		*out = new(int32)
+		**out = **in
+	}
+	if in.MinExecutors != nil {
+		in, out := &in.MinExecutors, &out.MinExecutors
+		*out = new(int32)
+		**out = **in
+	}
+	if in.MaxExecutors != nil {
+		in, out := &in.MaxExecutors, &out.MaxExecutors
+		*out = new(int32)
+		**out = **in
+	}
+	if in.ShuffleTrackingTimeout != nil {
+		in, out := &in.ShuffleTrackingTimeout, &out.ShuffleTrackingTimeout
+		*out = new(int64)
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DynamicAllocation.
+func (in *DynamicAllocation) DeepCopy() *DynamicAllocation {
+	if in == nil {
+		return nil
+	}
+	out := new(DynamicAllocation)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ExecutorSpec) DeepCopyInto(out *ExecutorSpec) {
 	*out = *in
@@ -667,6 +703,11 @@ func (in *SparkApplicationSpec) DeepCopyInto(out *SparkApplicationSpec) {
 		*out = new(SparkUIConfiguration)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.DynamicAllocation != nil {
+		in, out := &in.DynamicAllocation, &out.DynamicAllocation
+		*out = new(DynamicAllocation)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
diff --git a/pkg/batchscheduler/volcano/volcano_scheduler.go b/pkg/batchscheduler/volcano/volcano_scheduler.go
index d4f6a06..1e3c381 100644
--- a/pkg/batchscheduler/volcano/volcano_scheduler.go
+++ b/pkg/batchscheduler/volcano/volcano_scheduler.go
@@ -75,7 +75,14 @@ func (v *VolcanoBatchScheduler) DoBatchSchedulingOnSubmission(app *v1beta2.Spark
 func (v *VolcanoBatchScheduler) syncPodGroupInClientMode(app *v1beta2.SparkApplication) error {
 	// We only care about the executor pods in client mode
 	if _, ok := app.Spec.Executor.Annotations[v1beta1.KubeGroupNameAnnotationKey]; !ok {
-		if err := v.syncPodGroup(app, 1, getExecutorRequestResource(app)); err == nil {
+		var totalResource corev1.ResourceList
+		requestResources := app.Spec.BatchSchedulerOptions.Resources
+		if len(requestResources) > 0 {
+			totalResource = requestResources
+		} else {
+			totalResource = getExecutorRequestResource(app)
+		}
+		if err := v.syncPodGroup(app, 1, totalResource); err == nil {
 			app.Spec.Executor.Annotations[v1beta1.KubeGroupNameAnnotationKey] = v.getAppPodGroupName(app)
 		} else {
 			return err
@@ -90,6 +97,10 @@ func (v *VolcanoBatchScheduler) syncPodGroupInClusterMode(app *v1beta2.SparkAppl
 	if _, ok := app.Spec.Driver.Annotations[v1beta1.KubeGroupNameAnnotationKey]; !ok {
 		//Both driver and executor resource will be considered.
 		totalResource := sumResourceList([]corev1.ResourceList{getExecutorRequestResource(app), getDriverRequestResource(app)})
+		requestResources := app.Spec.BatchSchedulerOptions.Resources
+		if len(requestResources) > 0 {
+			totalResource = requestResources
+		}
 		if err := v.syncPodGroup(app, 1, totalResource); err == nil {
 			app.Spec.Executor.Annotations[v1beta1.KubeGroupNameAnnotationKey] = v.getAppPodGroupName(app)
 			app.Spec.Driver.Annotations[v1beta1.KubeGroupNameAnnotationKey] = v.getAppPodGroupName(app)
diff --git a/pkg/config/constants.go b/pkg/config/constants.go
index 07ba29b..85deffe 100644
--- a/pkg/config/constants.go
+++ b/pkg/config/constants.go
@@ -148,6 +148,24 @@ const (
 	SparkDriverKubernetesMaster = "spark.kubernetes.driver.master"
 	// SparkDriverServiceAnnotationKeyPrefix is the key prefix of annotations to be added to the driver service.
 	SparkDriverServiceAnnotationKeyPrefix = "spark.kubernetes.driver.service.annotation."
+	// SparkDynamicAllocationEnabled is the Spark configuration key for specifying if dynamic
+	// allocation is enabled or not.
+	SparkDynamicAllocationEnabled = "spark.dynamicAllocation.enabled"
+	// SparkDynamicAllocationShuffleTrackingEnabled is the Spark configuration key for
+	// specifying if shuffle data tracking is enabled.
+	SparkDynamicAllocationShuffleTrackingEnabled = "spark.dynamicAllocation.shuffleTracking.enabled"
+	// SparkDynamicAllocationShuffleTrackingTimeout is the Spark configuration key for specifying
+	// the shuffle tracking timeout in milliseconds if shuffle tracking is enabled.
+	SparkDynamicAllocationShuffleTrackingTimeout = "spark.dynamicAllocation.shuffleTracking.timeout"
+	// SparkDynamicAllocationInitialExecutors is the Spark configuration key for specifying
+	// the initial number of executors to request if dynamic allocation is enabled.
+	SparkDynamicAllocationInitialExecutors = "spark.dynamicAllocation.initialExecutors"
+	// SparkDynamicAllocationMinExecutors is the Spark configuration key for specifying the
+	// lower bound of the number of executors to request if dynamic allocation is enabled.
+	SparkDynamicAllocationMinExecutors = "spark.dynamicAllocation.minExecutors"
+	// SparkDynamicAllocationMaxExecutors is the Spark configuration key for specifying the
+	// upper bound of the number of executors to request if dynamic allocation is enabled.
+	SparkDynamicAllocationMaxExecutors = "spark.dynamicAllocation.maxExecutors"
 )
 
 const (
diff --git a/pkg/controller/sparkapplication/submission.go b/pkg/controller/sparkapplication/submission.go
index 8955e86..cebf3b0 100644
--- a/pkg/controller/sparkapplication/submission.go
+++ b/pkg/controller/sparkapplication/submission.go
@@ -140,11 +140,6 @@ func buildSubmissionCommandArgs(app *v1beta2.SparkApplication, driverPodName str
 		args = append(args, "--conf", fmt.Sprintf("spark.hadoop.%s=%s", key, value))
 	}
 
-	for key, value := range app.Spec.NodeSelector {
-		conf := fmt.Sprintf("%s%s=%s", config.SparkNodeSelectorKeyPrefix, key, value)
-		args = append(args, "--conf", conf)
-	}
-
 	// Add the driver and executor configuration options.
 	// Note that when the controller submits the application, it expects that all dependencies are local
 	// so init-container is not needed and therefore no init-container image needs to be specified.
@@ -163,6 +158,16 @@ func buildSubmissionCommandArgs(app *v1beta2.SparkApplication, driverPodName str
 		args = append(args, "--conf", option)
 	}
 
+	options = addDynamicAllocationConfOptions(app)
+	for _, option := range options {
+		args = append(args, "--conf", option)
+	}
+
+	for key, value := range app.Spec.NodeSelector {
+		conf := fmt.Sprintf("%s%s=%s", config.SparkNodeSelectorKeyPrefix, key, value)
+		args = append(args, "--conf", conf)
+	}
+
 	if app.Spec.Volumes != nil {
 		options, err = addLocalDirConfOptions(app)
 		if err != nil {
@@ -393,6 +398,36 @@ func addExecutorConfOptions(app *v1beta2.SparkApplication, submissionID string)
 	return executorConfOptions, nil
 }
 
+func addDynamicAllocationConfOptions(app *v1beta2.SparkApplication) []string {
+	if app.Spec.DynamicAllocation == nil {
+		return nil
+	}
+
+	dynamicAllocation := app.Spec.DynamicAllocation
+	if !dynamicAllocation.Enabled {
+		return nil
+	}
+
+	var options []string
+	options = append(options, fmt.Sprintf("%s=true", config.SparkDynamicAllocationEnabled))
+	// Turn on shuffle tracking if dynamic allocation is enabled.
+	options = append(options, fmt.Sprintf("%s=true", config.SparkDynamicAllocationShuffleTrackingEnabled))
+	if dynamicAllocation.InitialExecutors != nil {
+		options = append(options, fmt.Sprintf("%s=%d", config.SparkDynamicAllocationInitialExecutors, *dynamicAllocation.InitialExecutors))
+	}
+	if dynamicAllocation.MinExecutors != nil {
+		options = append(options, fmt.Sprintf("%s=%d", config.SparkDynamicAllocationMinExecutors, *dynamicAllocation.MinExecutors))
+	}
+	if dynamicAllocation.MaxExecutors != nil {
+		options = append(options, fmt.Sprintf("%s=%d", config.SparkDynamicAllocationMaxExecutors, *dynamicAllocation.MaxExecutors))
+	}
+	if dynamicAllocation.ShuffleTrackingTimeout != nil {
+		options = append(options, fmt.Sprintf("%s=%d", config.SparkDynamicAllocationShuffleTrackingTimeout, *dynamicAllocation.ShuffleTrackingTimeout))
+	}
+
+	return options
+}
+
 // addLocalDirConfOptions excludes local dir volumes, update SparkApplication and returns local dir config options
 func addLocalDirConfOptions(app *v1beta2.SparkApplication) ([]string, error) {
 	var localDirConfOptions []string
diff --git a/spark-docker/Dockerfile b/spark-docker/Dockerfile
index e6196e9..c52b48e 100644
--- a/spark-docker/Dockerfile
+++ b/spark-docker/Dockerfile
@@ -14,20 +14,30 @@
 # limitations under the License.
 #
 
-ARG SPARK_IMAGE=gcr.io/spark-operator/spark:v2.4.5
+ARG SPARK_IMAGE=gcr.io/spark-operator/spark:v3.0.0
 FROM ${SPARK_IMAGE}
 
+# Switch to user root so we can add addtional jars and configuration files.
+USER root
+
 # Setup dependencies for Google Cloud Storage access.
 RUN rm $SPARK_HOME/jars/guava-14.0.1.jar
 ADD https://repo1.maven.org/maven2/com/google/guava/guava/23.0/guava-23.0.jar $SPARK_HOME/jars
+RUN chmod 644 $SPARK_HOME/jars/guava-23.0.jar
 # Add the connector jar needed to access Google Cloud Storage using the Hadoop FileSystem API.
 ADD https://storage.googleapis.com/hadoop-lib/gcs/gcs-connector-latest-hadoop2.jar $SPARK_HOME/jars
+RUN chmod 644 $SPARK_HOME/jars/gcs-connector-latest-hadoop2.jar
 ADD https://storage.googleapis.com/spark-lib/bigquery/spark-bigquery-latest.jar $SPARK_HOME/jars
+RUN chmod 644 $SPARK_HOME/jars/spark-bigquery-latest.jar
 
 # Setup for the Prometheus JMX exporter.
-RUN mkdir -p /etc/metrics/conf
 # Add the Prometheus JMX exporter Java agent jar for exposing metrics sent to the JmxSink to Prometheus.
 ADD https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.11.0/jmx_prometheus_javaagent-0.11.0.jar /prometheus/
+RUN chmod 644 /prometheus/jmx_prometheus_javaagent-0.11.0.jar
+
+USER ${spark_uid}
+
+RUN mkdir -p /etc/metrics/conf
 COPY conf/metrics.properties /etc/metrics/conf
 COPY conf/prometheus.yaml /etc/metrics/conf
 


Test patch:
diff --git a/pkg/batchscheduler/volcano/volcano_scheduler_test.go b/pkg/batchscheduler/volcano/volcano_scheduler_test.go
index 7c5bfe4..a3d9cf7 100644
--- a/pkg/batchscheduler/volcano/volcano_scheduler_test.go
+++ b/pkg/batchscheduler/volcano/volcano_scheduler_test.go
@@ -19,7 +19,7 @@ package volcano
 import (
 	"testing"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 
 	"github.com/GoogleCloudPlatform/spark-on-k8s-operator/pkg/apis/sparkoperator.k8s.io/v1beta2"
diff --git a/pkg/controller/sparkapplication/submission_test.go b/pkg/controller/sparkapplication/submission_test.go
index 5a26d61..e00015b 100644
--- a/pkg/controller/sparkapplication/submission_test.go
+++ b/pkg/controller/sparkapplication/submission_test.go
@@ -18,18 +18,20 @@ package sparkapplication
 
 import (
 	"fmt"
-	"github.com/google/uuid"
 	"reflect"
 	"sort"
 	"strconv"
 	"testing"
 
+	"github.com/google/uuid"
+
 	"github.com/stretchr/testify/assert"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/GoogleCloudPlatform/spark-on-k8s-operator/pkg/apis/sparkoperator.k8s.io/v1beta2"
+	"github.com/GoogleCloudPlatform/spark-on-k8s-operator/pkg/config"
 )
 
 const (
@@ -504,3 +506,40 @@ func TestPopulateLabelsOverride_Driver_Executor(t *testing.T) {
 		t.Errorf("Executor labels: wanted %+q got %+q", expectedExecutorLabels, executorOptions)
 	}
 }
+
+func TestDynamicAllocationOptions(t *testing.T) {
+	app := &v1beta2.SparkApplication{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-test",
+			UID:  "spark-test-1",
+		},
+		Spec: v1beta2.SparkApplicationSpec{},
+	}
+	options := addDynamicAllocationConfOptions(app)
+	assert.Equal(t, 0, len(options))
+
+	app = &v1beta2.SparkApplication{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-test",
+			UID:  "spark-test-1",
+		},
+		Spec: v1beta2.SparkApplicationSpec{
+			DynamicAllocation: &v1beta2.DynamicAllocation{
+				Enabled:                true,
+				InitialExecutors:       int32ptr(2),
+				MinExecutors:           int32ptr(0),
+				MaxExecutors:           int32ptr(10),
+				ShuffleTrackingTimeout: int64ptr(6000000),
+			},
+		},
+	}
+
+	options = addDynamicAllocationConfOptions(app)
+	assert.Equal(t, 6, len(options))
+	assert.Equal(t, fmt.Sprintf("%s=true", config.SparkDynamicAllocationEnabled), options[0])
+	assert.Equal(t, fmt.Sprintf("%s=true", config.SparkDynamicAllocationShuffleTrackingEnabled), options[1])
+	assert.Equal(t, fmt.Sprintf("%s=2", config.SparkDynamicAllocationInitialExecutors), options[2])
+	assert.Equal(t, fmt.Sprintf("%s=0", config.SparkDynamicAllocationMinExecutors), options[3])
+	assert.Equal(t, fmt.Sprintf("%s=10", config.SparkDynamicAllocationMaxExecutors), options[4])
+	assert.Equal(t, fmt.Sprintf("%s=6000000", config.SparkDynamicAllocationShuffleTrackingTimeout), options[5])
+}
diff --git a/test/e2e/README.md b/test/e2e/README.md
index 4eaed1e..64859d2 100644
--- a/test/e2e/README.md
+++ b/test/e2e/README.md
@@ -12,7 +12,7 @@ Prerequisites:
 e2e tests are written as Go test. All go test techniques apply (e.g. picking what to run, timeout length). Let's say I want to run all tests in "test/e2e/":
 
 ```bash
-$ go test -v ./test/e2e/ --kubeconfig "$HOME/.kube/config" --operator-image=gcr.io/spark-operator/spark-operator:v2.4.5-v1beta2-latest
+$ go test -v ./test/e2e/ --kubeconfig "$HOME/.kube/config" --operator-image=gcr.io/spark-operator/spark-operator:v1beta2-1.2.0-3.0.0
 ```
 
 ### Available Tests
diff --git a/test/e2e/basic_test.go b/test/e2e/basic_test.go
index d74f82f..03a48e8 100644
--- a/test/e2e/basic_test.go
+++ b/test/e2e/basic_test.go
@@ -22,7 +22,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 
 	appFramework "github.com/GoogleCloudPlatform/spark-on-k8s-operator/test/e2e/framework"
@@ -70,3 +70,46 @@ func TestSubmitSparkPiYaml(t *testing.T) {
 	err = appFramework.DeleteSparkApplication(framework.SparkApplicationClient, appFramework.SparkTestNamespace, appName)
 	assert.Equal(t, nil, err)
 }
+
+func TestSubmitSparkPiCustomResourceYaml(t *testing.T) {
+	t.Parallel()
+
+	appName := "spark-pi-custom-resource"
+	sa, err := appFramework.MakeSparkApplicationFromYaml("../../examples/spark-pi-custom-resource.yaml")
+	assert.Equal(t, nil, err)
+
+	if appFramework.SparkTestNamespace != "" {
+		sa.ObjectMeta.Namespace = appFramework.SparkTestNamespace
+	}
+
+	if appFramework.SparkTestServiceAccount != "" {
+		sa.Spec.Driver.ServiceAccount = &appFramework.SparkTestServiceAccount
+	}
+
+	if appFramework.SparkTestImage != "" {
+		sa.Spec.Image = &appFramework.SparkTestImage
+	}
+
+	err = appFramework.CreateSparkApplication(framework.SparkApplicationClient, appFramework.SparkTestNamespace, sa)
+	assert.Equal(t, nil, err)
+
+	status := GetJobStatus(t, appName)
+
+	err = wait.Poll(INTERVAL, TIMEOUT, func() (done bool, err error) {
+		if status == "COMPLETED" {
+			return true, nil
+		}
+		status = GetJobStatus(t, appName)
+		return false, nil
+	})
+	assert.Equal(t, nil, err)
+
+	app, _ := appFramework.GetSparkApplication(framework.SparkApplicationClient, appFramework.SparkTestNamespace, appName)
+	podName := app.Status.DriverInfo.PodName
+	rawLogs, err := framework.KubeClient.CoreV1().Pods(appFramework.SparkTestNamespace).GetLogs(podName, &v1.PodLogOptions{}).Do().Raw()
+	assert.Equal(t, nil, err)
+	assert.NotEqual(t, -1, strings.Index(string(rawLogs), "Pi is roughly 3"))
+
+	err = appFramework.DeleteSparkApplication(framework.SparkApplicationClient, appFramework.SparkTestNamespace, appName)
+	assert.Equal(t, nil, err)
+}