Repository: kubeflow/spark-operator

Issue:
[BUG] SparkApplication in FAILING state has finish time
## Description
We are using spark operator  v1beta2-1.6.2-3.5.0 in production. We have a spark application with the below policy. 

```
  restartPolicy:
    onFailureRetryInterval: 10
    onSubmissionFailureRetries: 5
    onSubmissionFailureRetryInterval: 20
    type: Always
```

When driver pod failed for some reason, operator has the below logs 


```
UID:"263d8e51-4a32-4832-94b5-f73043e4bd69", APIVersion:"sparkoperator.k8s.io/v1beta2", ResourceVersion:"1343465049", FieldPath:""}): type: 'Warning' reason: 'SparkDriverFailed' Driver **-regular-driver failed
I0810 13:25:06.814383      10 controller.go:860] Update the status of SparkApplication namespace/**-regular from:
{
  "sparkApplicationId": "spark-5a8712b706984da5bd90ac3fa7f35b86",
  "submissionID": "9b80c355-fc31-4680-b3ce-7ce34f7c31ec",
  "lastSubmissionAttemptTime": "2024-08-10T10:26:46Z",
  "terminationTime": null,
  "driverInfo": {
    "webUIServiceName": "**-regular-ui-svc",
    "webUIPort": 4040,
    "webUIAddress": "10.204.49.228:0",
    "podName": "**-regular-driver"
  },
  "applicationState": {
    "state": "RUNNING"
  },
  "executorState": {
    "**-8f53b6913bd57dcf-exec-7": "UNKNOWN",
    "**-8f53b6913bd57dcf-exec-8": "FAILED",
    "**-8f53b6913bd57dcf-exec-9": "COMPLETED"
  },
  "executionAttempts": 63,
  "submissionAttempts": 1
}
to:
{
  "sparkApplicationId": "spark-5a8712b706984da5bd90ac3fa7f35b86",
  "submissionID": "9b80c355-fc31-4680-b3ce-7ce34f7c31ec",
  "lastSubmissionAttemptTime": "2024-08-10T10:26:46Z",
  "terminationTime": "2024-08-10T13:25:06Z",
  "driverInfo": {
    "webUIServiceName": "**-regular-ui-svc",
    "webUIPort": 4040,
    "webUIAddress": "10.204.49.228:0",
    "podName": "**-regular-driver"
  },
  "applicationState": {
    "state": "FAILING",
    "errorMessage": "driver container failed with ExitCode: 1, Reason: Error"
  },
  "executorState": {
    "**-8f53b6913bd57dcf-exec-8": "FAILED",
    "**-8f53b6913bd57dcf-exec-9": "COMPLETED"
  },
  "executionAttempts": 63,
  "submissionAttempts": 1
}
I0810 13:25:06.827513      10 metrics.go:125] Decrementing spark_app_running_count with labels map[app_type:Unknown] metricVal to 4
I0810 13:25:06.827552      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-4. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827564      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 29
I0810 13:25:06.827571      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-18. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827575      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 28
I0810 13:25:06.827579      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-19. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827586      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 27
I0810 13:25:06.827593      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-3. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827607      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 26
I0810 13:25:06.827613      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-35. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827622      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 25
I0810 13:25:06.827631      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-7. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827640      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 24
I0810 13:25:06.827647      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-1. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827654      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 23
I0810 13:25:06.827662      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-24. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827669      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 22
I0810 13:25:06.827675      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-40. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827679      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 21
I0810 13:25:06.827683      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-5. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827689      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 20
I0810 13:25:06.827693      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-2. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827697      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 19
I0810 13:25:06.827701      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-32. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827707      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 18
I0810 13:25:06.827711      10 sparkapp_metrics.go:287] Exporting Metrics for Executor **-8f53b6913bd57dcf-exec-38. OldState: UNKNOWN NewState: FAILED
I0810 13:25:06.827717      10 metrics.go:125] Decrementing spark_app_executor_running_count with labels map[app_type:Unknown] metricVal to 17
I0810 13:25:06.827722      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:25:06.827792      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:25:06.827814      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:25:06.827855      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:25:07.452988      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:25:07.453037      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:25:07.453109      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:25:07.909475      10 spark_pod_eventhandler.go:58] Pod **-regular-driver updated in namespace namespace.
I0810 13:25:07.909518      10 spark_pod_eventhandler.go:95] Enqueuing SparkApplication namespace/**-regular for app update processing.
I0810 13:25:07.909544      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:25:07.909634      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:25:07.969839      10 spark_pod_eventhandler.go:58] Pod **-regular-driver updated in namespace namespace.
I0810 13:25:07.969874      10 spark_pod_eventhandler.go:95] Enqueuing SparkApplication namespace/**-regular for app update processing.
I0810 13:25:07.969899      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:25:07.969972      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:25:08.822351      10 spark_pod_eventhandler.go:58] Pod **-regular-driver updated in namespace namespace.
I0810 13:25:08.822380      10 spark_pod_eventhandler.go:95] Enqueuing SparkApplication namespace/**-regular for app update processing.
I0810 13:25:08.822402      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:25:08.822479      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:25:37.453315      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:25:37.453383      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:25:37.453478      10 controller.go:274] Ending processing key: "namespace/**-regular"


I0810 13:26:07.453988      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:26:07.454063      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:26:07.454133      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:26:37.454371      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:26:37.454424      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:26:37.454507      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:27:07.455311      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:27:07.455388      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:27:07.455463      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:27:37.455431      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:27:37.455488      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:27:37.455562      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:27:50.436345      10 spark_pod_eventhandler.go:58] Pod **-regular-driver updated in namespace namespace.
I0810 13:27:50.436379      10 spark_pod_eventhandler.go:95] Enqueuing SparkApplication namespace/**-regular for app update processing.
I0810 13:27:50.436405      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:27:50.436492      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:27:50.440231      10 spark_pod_eventhandler.go:77] Pod **-regular-driver deleted in namespace namespace.
I0810 13:27:50.440254      10 spark_pod_eventhandler.go:95] Enqueuing SparkApplication namespace/**-regular for app update processing.
I0810 13:27:50.440271      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:27:50.440324      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:28:07.456297      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:28:07.456357      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:28:07.456440      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:28:37.456984      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:28:37.457038      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:28:37.457124      10 controller.go:274] Ending processing key: "namespace/**-regular"
I0810 13:29:07.457237      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:29:07.457300      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:29:07.457400      10 controller.go:274] Ending processing key: "namespace/**-regular"


I0810 13:29:37.458045      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:29:37.458105      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:29:37.458285      10 controller.go:274] Ending processing key: "namespace/**-regular"


I0810 13:30:07.458243      10 controller.go:227] SparkApplication namespace/**-regular was updated, enqueuing it
I0810 13:30:07.458322      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:30:07.458431      10 controller.go:274] Ending processing key: "namespace/**-regular"

```
driver pod is in error state and the sparkapplication state is 

```
NAME           STATUS    ATTEMPTS   START                                FINISH                              AGE
**-regular    FAILING   63                 2024-08-10T10:26:46Z   2024-08-10T13:25:06Z   4d5h
```

only after deleting the sparkapplication manually, operator has the logs and started the spark application

```
I0810 13:30:31.202106      10 controller.go:896] Deleting pod **-regular-driver in namespace namespace
I0810 13:30:31.205664      10 controller.go:904] Deleting Spark UI Service **-regular-ui-svc in namespace namespace
I0810 13:30:31.224970      10 event.go:364] Event(v1.ObjectReference{Kind:"SparkApplication", Namespace:"namespace", Name:"**-regular", UID:"263d8e51-4a32-4832-94b5-f73043e4bd69", APIVersion:"sparkoperator.k8s.io/v1beta2", ResourceVersion:"1343488188", FieldPath:""}): type: 'Normal' reason: 'SparkApplicationDeleted' SparkApplication **-regular was deleted
I0810 13:30:35.985819      10 controller.go:188] SparkApplication namespace/**-regular was added, enqueuing it for submission
I0810 13:30:35.985869      10 controller.go:267] Starting processing key: "namespace/**-regular"
I0810 13:30:35.985929      10 driveringress.go:287] Creating a service **-regular-ui-svc for the Driver Ingress for application **-regular
I0
```

How to make sure that my sparkapplication gets restarted when the driver failed, this is happening regularly.




- [x] ✋ I have searched the open/closed issues and my issue is not listed.

## Reproduction Code [Required]

<!-- REQUIRED -->

Submit the sparkApplication to spark operator with 

```
  restartPolicy:
    onFailureRetryInterval: 10
    onSubmissionFailureRetries: 5
    onSubmissionFailureRetryInterval: 20
    type: Always
```

when the driver pod fails, sparkapplication fails and doesn't submit again

## Expected behavior

Spark application should be restarted 

## Actual behavior

SparkApplication has finish time and is in Failing State. 

### Terminal Output Screenshot(s)

<!-- Optional but helpful -->


## Environment & Versions

- Spark Operator App version: v1beta2-1.6.2-3.5.0
- Helm Chart Version: 1.4.5
- Kubernetes Version:  v1.28.7-gke.1026001
- Apache Spark version:  3.4.0

## Additional context

<!-- Add any other context about the problem here -->




Hints from issue comments:
@TheDevilDan: I set the restartPolicy to type: Always, but it never restarts on error, OOM Kill.

And when I edit the driver configuration in Kubernetes, I see that the policy applied is: NEVER

Whereas in the SparkApplication, it's Always!

Tip: To restart the application, sometimes go to the shell in the pod, do an exit 1 and it restarts, if it remains in error, delete the pod and it restarts... Otherwise, delete the SparkAplication and Upgrade the chart if necessary / re-apply the yaml.

On spark application : restartPolicy type: Always
![image](https://github.com/user-attachments/assets/39b51695-c041-407b-b00c-d7f7d3c73afa)

On driver config : restartPolicy type: Never (automatic definition by spark operator bug ?)
![image](https://github.com/user-attachments/assets/b891eb5e-e32d-41b0-9c7c-89b775be89ea)
@github-actions: This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.@github-actions: This issue has been automatically closed because it has not had recent activity. Please comment "/reopen" to reopen it.@ChenYi015: > I set the restartPolicy to type: Always, but it never restarts on error, OOM Kill.
> 
> And when I edit the driver configuration in Kubernetes, I see that the policy applied is: NEVER
> 
> Whereas in the SparkApplication, it's Always!

The restartPolicy of SparkApplication does not mean the restartPolicy of Spark pods. The restartPolicy of SparkApplication will be handled by Spark operator at application level.

Solution patch:
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
index 532cd98..e428fe0 100644
--- a/.github/workflows/integration.yaml
+++ b/.github/workflows/integration.yaml
@@ -209,6 +209,9 @@ jobs:
 
   e2e-test:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        k8s_version: [v1.28.13, v1.29.8, v1.30.4, v1.31.1]
     steps:
       - name: Checkout source code
         uses: actions/checkout@v4
@@ -221,7 +224,7 @@ jobs:
           go-version-file: go.mod
 
       - name: Create a Kind cluster
-        run: make kind-create-cluster
+        run: make kind-create-cluster KIND_K8S_VERSION=${{ matrix.k8s_version }}
 
       - name: Build and load image to Kind cluster
         run: |
diff --git a/.github/workflows/trivy-image-scanning.yaml b/.github/workflows/trivy-image-scanning.yaml
index a7bd2a9..e0ba85c 100644
--- a/.github/workflows/trivy-image-scanning.yaml
+++ b/.github/workflows/trivy-image-scanning.yaml
@@ -15,7 +15,7 @@ jobs:
         run: make print-IMAGE >> $GITHUB_ENV
 
       - name: trivy scan for github security tab
-        uses: aquasecurity/trivy-action@0.27.0
+        uses: aquasecurity/trivy-action@0.28.0
         with:
           image-ref: '${{ env.IMAGE }}'
           format: 'sarif'
diff --git a/Makefile b/Makefile
index 901dbfc..3283f2a 100644
--- a/Makefile
+++ b/Makefile
@@ -52,6 +52,7 @@ LOCALBIN ?= $(shell pwd)/bin
 KUSTOMIZE_VERSION ?= v5.4.1
 CONTROLLER_TOOLS_VERSION ?= v0.15.0
 KIND_VERSION ?= v0.23.0
+KIND_K8S_VERSION ?= v1.29.3
 ENVTEST_VERSION ?= release-0.18
 # ENVTEST_K8S_VERSION refers to the version of kubebuilder assets to be downloaded by envtest binary.
 ENVTEST_K8S_VERSION ?= 1.29.3
@@ -249,7 +250,12 @@ endif
 .PHONY: kind-create-cluster
 kind-create-cluster: kind ## Create a kind cluster for integration tests.
 	if ! $(KIND) get clusters 2>/dev/null | grep -q "^$(KIND_CLUSTER_NAME)$$"; then \
-		$(KIND) create cluster --name $(KIND_CLUSTER_NAME) --config $(KIND_CONFIG_FILE) --kubeconfig $(KIND_KUBE_CONFIG) --wait=1m; \
+		$(KIND) create cluster \
+			--name $(KIND_CLUSTER_NAME) \
+			--config $(KIND_CONFIG_FILE) \
+			--image kindest/node:$(KIND_K8S_VERSION) \
+			--kubeconfig $(KIND_KUBE_CONFIG) \
+			--wait=1m; \
 	fi
 
 .PHONY: kind-load-image
diff --git a/charts/spark-operator-chart/README.md b/charts/spark-operator-chart/README.md
index 9c945d3..1df28cd 100644
--- a/charts/spark-operator-chart/README.md
+++ b/charts/spark-operator-chart/README.md
@@ -90,6 +90,7 @@ See [helm uninstall](https://helm.sh/docs/helm/helm_uninstall) for command docum
 | controller.uiService.enable | bool | `true` | Specifies whether to create service for Spark web UI. |
 | controller.uiIngress.enable | bool | `false` | Specifies whether to create ingress for Spark web UI. `controller.uiService.enable` must be `true` to enable ingress. |
 | controller.uiIngress.urlFormat | string | `""` | Ingress URL format. Required if `controller.uiIngress.enable` is true. |
+| controller.uiIngress.ingressClassName | string | `""` | Optionally set the ingressClassName. |
 | controller.batchScheduler.enable | bool | `false` | Specifies whether to enable batch scheduler for spark jobs scheduling. If enabled, users can specify batch scheduler name in spark application. |
 | controller.batchScheduler.kubeSchedulerNames | list | `[]` | Specifies a list of kube-scheduler names for scheduling Spark pods. |
 | controller.batchScheduler.default | string | `""` | Default batch scheduler to be used if not specified by the user. If specified, this value must be either "volcano" or "yunikorn". Specifying any other value will cause the controller to error on startup. |
diff --git a/charts/spark-operator-chart/ci/kind-config.yaml b/charts/spark-operator-chart/ci/kind-config.yaml
index 4e8cae8..cebee6d 100644
--- a/charts/spark-operator-chart/ci/kind-config.yaml
+++ b/charts/spark-operator-chart/ci/kind-config.yaml
@@ -2,6 +2,4 @@ kind: Cluster
 apiVersion: kind.x-k8s.io/v1alpha4
 nodes:
   - role: control-plane
-    image: kindest/node:v1.29.2
   - role: worker
-    image: kindest/node:v1.29.2
diff --git a/charts/spark-operator-chart/templates/controller/deployment.yaml b/charts/spark-operator-chart/templates/controller/deployment.yaml
index 1efd126..2a1fd5f 100644
--- a/charts/spark-operator-chart/templates/controller/deployment.yaml
+++ b/charts/spark-operator-chart/templates/controller/deployment.yaml
@@ -69,6 +69,9 @@ spec:
         {{- with .Values.controller.uiIngress.urlFormat }}
         - --ingress-url-format={{ . }}
         {{- end }}
+        {{- with .Values.controller.uiIngress.ingressClassName }}
+        - --ingress-class-name={{ . }}
+        {{- end }}
         {{- end }}
         {{- if .Values.controller.batchScheduler.enable }}
         - --enable-batch-scheduler=true
diff --git a/charts/spark-operator-chart/values.yaml b/charts/spark-operator-chart/values.yaml
index 75cf5ca..672d622 100644
--- a/charts/spark-operator-chart/values.yaml
+++ b/charts/spark-operator-chart/values.yaml
@@ -65,6 +65,8 @@ controller:
     # -- Ingress URL format.
     # Required if `controller.uiIngress.enable` is true.
     urlFormat: ""
+    # -- Optionally set the ingressClassName.
+    ingressClassName: ""
 
   batchScheduler:
     # -- Specifies whether to enable batch scheduler for spark jobs scheduling.
diff --git a/go.mod b/go.mod
index c3133ce..d5e600b 100644
--- a/go.mod
+++ b/go.mod
@@ -3,16 +3,16 @@ module github.com/kubeflow/spark-operator
 go 1.23.1
 
 require (
-	cloud.google.com/go/storage v1.44.0
+	cloud.google.com/go/storage v1.45.0
 	github.com/aws/aws-sdk-go-v2 v1.32.2
-	github.com/aws/aws-sdk-go-v2/config v1.27.43
-	github.com/aws/aws-sdk-go-v2/service/s3 v1.65.3
+	github.com/aws/aws-sdk-go-v2/config v1.28.0
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.66.0
 	github.com/golang/glog v1.2.2
 	github.com/google/uuid v1.6.0
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/onsi/ginkgo/v2 v2.20.2
 	github.com/onsi/gomega v1.34.2
-	github.com/prometheus/client_golang v1.20.4
+	github.com/prometheus/client_golang v1.20.5
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/spf13/cobra v1.8.1
 	github.com/spf13/viper v1.19.0
@@ -21,10 +21,10 @@ require (
 	gocloud.dev v0.40.0
 	golang.org/x/net v0.30.0
 	golang.org/x/time v0.7.0
-	helm.sh/helm/v3 v3.16.1
-	k8s.io/api v0.31.0
-	k8s.io/apiextensions-apiserver v0.31.0
-	k8s.io/apimachinery v0.31.0
+	helm.sh/helm/v3 v3.16.2
+	k8s.io/api v0.31.1
+	k8s.io/apiextensions-apiserver v0.31.1
+	k8s.io/apimachinery v0.31.1
 	k8s.io/client-go v1.5.2
 	k8s.io/kubernetes v1.30.2
 	k8s.io/utils v0.0.0-20240711033017-18e509b52bc8
@@ -218,12 +218,12 @@ require (
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/apiserver v0.31.0 // indirect
-	k8s.io/cli-runtime v0.31.0 // indirect
-	k8s.io/component-base v0.31.0 // indirect
+	k8s.io/apiserver v0.31.1 // indirect
+	k8s.io/cli-runtime v0.31.1 // indirect
+	k8s.io/component-base v0.31.1 // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20240709000822-3c01b740850f // indirect
-	k8s.io/kubectl v0.31.0 // indirect
+	k8s.io/kubectl v0.31.1 // indirect
 	oras.land/oras-go v1.2.5 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/kustomize/api v0.17.2 // indirect
diff --git a/go.sum b/go.sum
index ef603e9..496567c 100644
--- a/go.sum
+++ b/go.sum
@@ -17,8 +17,8 @@ cloud.google.com/go/longrunning v0.6.1 h1:lOLTFxYpr8hcRtcwWir5ITh1PAKUD/sG2lKrTS
 cloud.google.com/go/longrunning v0.6.1/go.mod h1:nHISoOZpBcmlwbJmiVk5oDRz0qG/ZxPynEGs1iZ79s0=
 cloud.google.com/go/monitoring v1.21.0 h1:EMc0tB+d3lUewT2NzKC/hr8cSR9WsUieVywzIHetGro=
 cloud.google.com/go/monitoring v1.21.0/go.mod h1:tuJ+KNDdJbetSsbSGTqnaBvbauS5kr3Q/koy3Up6r+4=
-cloud.google.com/go/storage v1.44.0 h1:abBzXf4UJKMmQ04xxJf9dYM/fNl24KHoTuBjyJDX2AI=
-cloud.google.com/go/storage v1.44.0/go.mod h1:wpPblkIuMP5jCB/E48Pz9zIo2S/zD8g+ITmxKkPCITE=
+cloud.google.com/go/storage v1.45.0 h1:5av0QcIVj77t+44mV4gffFC/LscFRUhto6UBMB5SimM=
+cloud.google.com/go/storage v1.45.0/go.mod h1:wpPblkIuMP5jCB/E48Pz9zIo2S/zD8g+ITmxKkPCITE=
 cloud.google.com/go/trace v1.11.0 h1:UHX6cOJm45Zw/KIbqHe4kII8PupLt/V5tscZUkeiJVI=
 cloud.google.com/go/trace v1.11.0/go.mod h1:Aiemdi52635dBR7o3zuc9lLjXo3BwGaChEjCa3tJNmM=
 dario.cat/mergo v1.0.1 h1:Ra4+bf83h2ztPIQYNP99R6m+Y7KfnARDfID+a+vLl4s=
@@ -70,8 +70,8 @@ github.com/aws/aws-sdk-go-v2 v1.32.2 h1:AkNLZEyYMLnx/Q/mSKkcMqwNFXMAvFto9bNsHqcT
 github.com/aws/aws-sdk-go-v2 v1.32.2/go.mod h1:2SK5n0a2karNTv5tbP1SjsX0uhttou00v/HpXKM1ZUo=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.6 h1:pT3hpW0cOHRJx8Y0DfJUEQuqPild8jRGmSFmBgvydr0=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.6/go.mod h1:j/I2++U0xX+cr44QjHay4Cvxj6FUbnxrgmqN3H1jTZA=
-github.com/aws/aws-sdk-go-v2/config v1.27.43 h1:p33fDDihFC390dhhuv8nOmX419wjOSDQRb+USt20RrU=
-github.com/aws/aws-sdk-go-v2/config v1.27.43/go.mod h1:pYhbtvg1siOOg8h5an77rXle9tVG8T+BWLWAo7cOukc=
+github.com/aws/aws-sdk-go-v2/config v1.28.0 h1:FosVYWcqEtWNxHn8gB/Vs6jOlNwSoyOCA/g/sxyySOQ=
+github.com/aws/aws-sdk-go-v2/config v1.28.0/go.mod h1:pYhbtvg1siOOg8h5an77rXle9tVG8T+BWLWAo7cOukc=
 github.com/aws/aws-sdk-go-v2/credentials v1.17.41 h1:7gXo+Axmp+R4Z+AK8YFQO0ZV3L0gizGINCOWxSLY9W8=
 github.com/aws/aws-sdk-go-v2/credentials v1.17.41/go.mod h1:u4Eb8d3394YLubphT4jLEwN1rLNq2wFOlT6OuxFwPzU=
 github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.17 h1:TMH3f/SCAWdNtXXVPPu5D6wrr4G5hI1rAxbcocKfC7Q=
@@ -94,8 +94,8 @@ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.2 h1:s7NA1SOw8
 github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.2/go.mod h1:fnjjWyAW/Pj5HYOxl9LJqWtEwS7W2qgcRLWP+uWbss0=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.2 h1:t7iUP9+4wdc5lt3E41huP+GvQZJD38WLsgVp4iOtAjg=
 github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.18.2/go.mod h1:/niFCtmuQNxqx9v8WAPq5qh7EH25U4BF6tjoyq9bObM=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.65.3 h1:xxHGZ+wUgZNACQmxtdvP5tgzfsxGS3vPpTP5Hy3iToE=
-github.com/aws/aws-sdk-go-v2/service/s3 v1.65.3/go.mod h1:cB6oAuus7YXRZhWCc1wIwPywwZ1XwweNp2TVAEGYeB8=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.66.0 h1:xA6XhTF7PE89BCNHJbQi8VvPzcgMtmGC5dr8S8N7lHk=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.66.0/go.mod h1:cB6oAuus7YXRZhWCc1wIwPywwZ1XwweNp2TVAEGYeB8=
 github.com/aws/aws-sdk-go-v2/service/sso v1.24.2 h1:bSYXVyUzoTHoKalBmwaZxs97HU9DWWI3ehHSAMa7xOk=
 github.com/aws/aws-sdk-go-v2/service/sso v1.24.2/go.mod h1:skMqY7JElusiOUjMJMOv1jJsP7YUg7DrhgqZZWuzu1U=
 github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.2 h1:AhmO1fHINP9vFYUE0LHzCWg/LfUWUF+zFPEcY9QXb7o=
@@ -420,8 +420,8 @@ github.com/poy/onpar v1.1.2/go.mod h1:6X8FLNoxyr9kkmnlqpK6LSoiOtrO6MICtWwEuWkLjz
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
 github.com/prometheus/client_golang v1.1.0/go.mod h1:I1FGZT9+L76gKKOs5djB6ezCbFQP1xR9D75/vuwEF3g=
-github.com/prometheus/client_golang v1.20.4 h1:Tgh3Yr67PaOv/uTqloMsCEdeuFTatm5zIq5+qNN23vI=
-github.com/prometheus/client_golang v1.20.4/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE=
+github.com/prometheus/client_golang v1.20.5 h1:cxppBPuYhUnsO6yo/aoRol4L7q7UFfdm+bR9r+8l63Y=
+github.com/prometheus/client_golang v1.20.5/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
@@ -705,8 +705,8 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gotest.tools/v3 v3.4.0 h1:ZazjZUfuVeZGLAmlKKuyv3IKP5orXcwtOwDQH6YVr6o=
 gotest.tools/v3 v3.4.0/go.mod h1:CtbdzLSsqVhDgMtKsx03ird5YTGB3ar27v0u/yKBW5g=
-helm.sh/helm/v3 v3.16.1 h1:cER6tI/8PgUAsaJaQCVBUg3VI9KN4oVaZJgY60RIc0c=
-helm.sh/helm/v3 v3.16.1/go.mod h1:r+xBHHP20qJeEqtvBXMf7W35QDJnzY/eiEBzt+TfHps=
+helm.sh/helm/v3 v3.16.2 h1:Y9v7ry+ubQmi+cb5zw1Llx8OKHU9Hk9NQ/+P+LGBe2o=
+helm.sh/helm/v3 v3.16.2/go.mod h1:SyTXgKBjNqi2NPsHCW5dDAsHqvGIu0kdNYNH9gQaw70=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 k8s.io/api v0.29.3 h1:2ORfZ7+bGC3YJqGpV0KSDDEVf8hdGQ6A03/50vj8pmw=
diff --git a/internal/controller/sparkapplication/controller.go b/internal/controller/sparkapplication/controller.go
index 3b96a31..7a707df 100644
--- a/internal/controller/sparkapplication/controller.go
+++ b/internal/controller/sparkapplication/controller.go
@@ -172,6 +172,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		return ctrl.Result{Requeue: true}, err
 	}
 	logger.Info("Reconciling SparkApplication", "name", app.Name, "namespace", app.Namespace, "state", app.Status.AppState.State)
+	defer logger.Info("Finished reconciling SparkApplication", "name", app.Name, "namespace", app.Namespace)
 
 	// Check if the spark application is being deleted
 	if !app.DeletionTimestamp.IsZero() {
@@ -259,17 +260,7 @@ func (r *Reconciler) reconcileNewSparkApplication(ctx context.Context, req ctrl.
 			}
 			app := old.DeepCopy()
 
-			if err := r.submitSparkApplication(app); err != nil {
-				logger.Error(err, "Failed to submit SparkApplication", "name", app.Name, "namespace", app.Namespace)
-				app.Status = v1beta2.SparkApplicationStatus{
-					AppState: v1beta2.ApplicationState{
-						State:        v1beta2.ApplicationStateFailedSubmission,
-						ErrorMessage: err.Error(),
-					},
-					SubmissionAttempts:        app.Status.SubmissionAttempts + 1,
-					LastSubmissionAttemptTime: metav1.Now(),
-				}
-			}
+			_ = r.submitSparkApplication(app)
 			if err := r.updateSparkApplicationStatus(ctx, app); err != nil {
 				return err
 			}
@@ -315,6 +306,9 @@ func (r *Reconciler) reconcileSubmittedSparkApplication(ctx context.Context, req
 
 func (r *Reconciler) reconcileFailedSubmissionSparkApplication(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	key := req.NamespacedName
+
+	var result ctrl.Result
+
 	retryErr := retry.RetryOnConflict(
 		retry.DefaultRetry,
 		func() error {
@@ -328,15 +322,22 @@ func (r *Reconciler) reconcileFailedSubmissionSparkApplication(ctx context.Conte
 			app := old.DeepCopy()
 
 			if util.ShouldRetry(app) {
-				if isNextRetryDue(app) {
+				timeUntilNextRetryDue, err := util.TimeUntilNextRetryDue(app)
+				if err != nil {
+					return err
+				}
+				if timeUntilNextRetryDue <= 0 {
 					if r.validateSparkResourceDeletion(ctx, app) {
 						_ = r.submitSparkApplication(app)
 					} else {
 						if err := r.deleteSparkResources(ctx, app); err != nil {
 							logger.Error(err, "failed to delete resources associated with SparkApplication", "name", app.Name, "namespace", app.Namespace)
 						}
-						return err
+						return fmt.Errorf("resources associated with SparkApplication name: %s namespace: %s, needed to be deleted", app.Name, app.Namespace)
 					}
+				} else {
+					// If we're waiting before retrying then reconcile will not modify anything, so we need to requeue.
+					result.RequeueAfter = timeUntilNextRetryDue
 				}
 			} else {
 				app.Status.AppState.State = v1beta2.ApplicationStateFailed
@@ -352,9 +353,9 @@ func (r *Reconciler) reconcileFailedSubmissionSparkApplication(ctx context.Conte
 	)
 	if retryErr != nil {
 		logger.Error(retryErr, "Failed to reconcile SparkApplication", "name", key.Name, "namespace", key.Namespace)
-		return ctrl.Result{}, retryErr
+		return result, retryErr
 	}
-	return ctrl.Result{}, nil
+	return result, nil
 }
 
 func (r *Reconciler) reconcileRunningSparkApplication(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -408,9 +409,7 @@ func (r *Reconciler) reconcilePendingRerunSparkApplication(ctx context.Context,
 				logger.Info("Successfully deleted resources associated with SparkApplication", "name", app.Name, "namespace", app.Namespace, "state", app.Status.AppState.State)
 				r.recordSparkApplicationEvent(app)
 				r.resetSparkApplicationStatus(app)
-				if err = r.submitSparkApplication(app); err != nil {
-					logger.Error(err, "Failed to run spark-submit", "name", app.Name, "namespace", app.Namespace, "state", app.Status.AppState.State)
-				}
+				_ = r.submitSparkApplication(app)
 			}
 			if err := r.updateSparkApplicationStatus(ctx, app); err != nil {
 				return err
@@ -497,6 +496,9 @@ func (r *Reconciler) reconcileSucceedingSparkApplication(ctx context.Context, re
 
 func (r *Reconciler) reconcileFailingSparkApplication(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	key := req.NamespacedName
+
+	var result ctrl.Result
+
 	retryErr := retry.RetryOnConflict(
 		retry.DefaultRetry,
 		func() error {
@@ -510,12 +512,19 @@ func (r *Reconciler) reconcileFailingSparkApplication(ctx context.Context, req c
 			app := old.DeepCopy()
 
 			if util.ShouldRetry(app) {
-				if isNextRetryDue(app) {
+				timeUntilNextRetryDue, err := util.TimeUntilNextRetryDue(app)
+				if err != nil {
+					return err
+				}
+				if timeUntilNextRetryDue <= 0 {
 					if err := r.deleteSparkResources(ctx, app); err != nil {
 						logger.Error(err, "failed to delete spark resources", "name", app.Name, "namespace", app.Namespace)
 						return err
 					}
 					app.Status.AppState.State = v1beta2.ApplicationStatePendingRerun
+				} else {
+					// If we're waiting before retrying then reconcile will not modify anything, so we need to requeue.
+					result.RequeueAfter = timeUntilNextRetryDue
 				}
 			} else {
 				app.Status.AppState.State = v1beta2.ApplicationStateFailed
@@ -528,9 +537,9 @@ func (r *Reconciler) reconcileFailingSparkApplication(ctx context.Context, req c
 	)
 	if retryErr != nil {
 		logger.Error(retryErr, "Failed to reconcile SparkApplication", "name", key.Name, "namespace", key.Namespace)
-		return ctrl.Result{}, retryErr
+		return result, retryErr
 	}
-	return ctrl.Result{}, nil
+	return result, nil
 }
 
 func (r *Reconciler) reconcileCompletedSparkApplication(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -632,8 +641,28 @@ func (r *Reconciler) getSparkApplication(key types.NamespacedName) (*v1beta2.Spa
 }
 
 // submitSparkApplication creates a new submission for the given SparkApplication and submits it using spark-submit.
-func (r *Reconciler) submitSparkApplication(app *v1beta2.SparkApplication) error {
+func (r *Reconciler) submitSparkApplication(app *v1beta2.SparkApplication) (submitErr error) {
 	logger.Info("Submitting SparkApplication", "name", app.Name, "namespace", app.Namespace, "state", app.Status.AppState.State)
+	// SubmissionID must be set before creating any resources to ensure all the resources are labeled.
+	app.Status.SubmissionID = uuid.New().String()
+	app.Status.LastSubmissionAttemptTime = metav1.Now()
+	app.Status.SubmissionAttempts = app.Status.SubmissionAttempts + 1
+
+	defer func() {
+		if submitErr == nil {
+			app.Status.AppState = v1beta2.ApplicationState{
+				State: v1beta2.ApplicationStateSubmitted,
+			}
+			app.Status.ExecutionAttempts = app.Status.ExecutionAttempts + 1
+		} else {
+			logger.Info("Failed to submit SparkApplication", "name", app.Name, "namespace", app.Namespace, "state", app.Status.AppState.State, "error", submitErr)
+			app.Status.AppState = v1beta2.ApplicationState{
+				State:        v1beta2.ApplicationStateFailedSubmission,
+				ErrorMessage: submitErr.Error(),
+			}
+		}
+		r.recordSparkApplicationEvent(app)
+	}()
 
 	if util.PrometheusMonitoringEnabled(app) {
 		logger.Info("Configure Prometheus monitoring for SparkApplication", "name", app.Name, "namespace", app.Namespace)
@@ -709,7 +738,6 @@ func (r *Reconciler) submitSparkApplication(app *v1beta2.SparkApplication) error
 
 	driverPodName := util.GetDriverPodName(app)
 	app.Status.DriverInfo.PodName = driverPodName
-	app.Status.SubmissionID = uuid.New().String()
 	sparkSubmitArgs, err := buildSparkSubmitArgs(app)
 	if err != nil {
 		return fmt.Errorf("failed to build spark-submit arguments: %v", err)
@@ -717,44 +745,12 @@ func (r *Reconciler) submitSparkApplication(app *v1beta2.SparkApplication) error
 
 	// Try submitting the application by running spark-submit.
 	logger.Info("Running spark-submit for SparkApplication", "name", app.Name, "namespace", app.Namespace, "arguments", sparkSubmitArgs)
-	submitted, err := runSparkSubmit(newSubmission(sparkSubmitArgs, app))
-	if err != nil {
-		r.recordSparkApplicationEvent(app)
+	if err := runSparkSubmit(newSubmission(sparkSubmitArgs, app)); err != nil {
 		return fmt.Errorf("failed to run spark-submit: %v", err)
 	}
-	if !submitted {
-		// The application may not have been submitted even if err == nil, e.g., when some
-		// state update caused an attempt to re-submit the application, in which case no
-		// error gets returned from runSparkSubmit. If this is the case, we simply return.
-		return nil
-	}
-
-	app.Status.AppState = v1beta2.ApplicationState{
-		State: v1beta2.ApplicationStateSubmitted,
-	}
-	app.Status.SubmissionAttempts = app.Status.SubmissionAttempts + 1
-	app.Status.ExecutionAttempts = app.Status.ExecutionAttempts + 1
-	app.Status.LastSubmissionAttemptTime = metav1.Now()
-	r.recordSparkApplicationEvent(app)
 	return nil
 }
 
-// Helper func to determine if the next retry the SparkApplication is due now.
-func isNextRetryDue(app *v1beta2.SparkApplication) bool {
-	retryInterval := app.Spec.RestartPolicy.OnFailureRetryInterval
-	attemptsDone := app.Status.SubmissionAttempts
-	lastEventTime := app.Status.LastSubmissionAttemptTime
-	if retryInterval == nil || lastEventTime.IsZero() || attemptsDone <= 0 {
-		return false
-	}
-
-	// Retry if we have waited at-least equal to attempts*RetryInterval since we do a linear back-off.
-	interval := time.Duration(*retryInterval) * time.Second * time.Duration(attemptsDone)
-	currentTime := time.Now()
-	logger.Info(fmt.Sprintf("currentTime is %v, interval is %v", currentTime, interval))
-	return currentTime.After(lastEventTime.Add(interval))
-}
-
 // updateDriverState finds the driver pod of the application
 // and updates the driver state based on the current phase of the pod.
 func (r *Reconciler) updateDriverState(_ context.Context, app *v1beta2.SparkApplication) error {
diff --git a/internal/controller/sparkapplication/submission.go b/internal/controller/sparkapplication/submission.go
index ef9485a..66e4a0b 100644
--- a/internal/controller/sparkapplication/submission.go
+++ b/internal/controller/sparkapplication/submission.go
@@ -43,10 +43,10 @@ func newSubmission(args []string, app *v1beta2.SparkApplication) *submission {
 	}
 }
 
-func runSparkSubmit(submission *submission) (bool, error) {
+func runSparkSubmit(submission *submission) error {
 	sparkHome, present := os.LookupEnv(common.EnvSparkHome)
 	if !present {
-		return false, fmt.Errorf("env %s is not specified", common.EnvSparkHome)
+		return fmt.Errorf("env %s is not specified", common.EnvSparkHome)
 	}
 	command := filepath.Join(sparkHome, "bin", "spark-submit")
 	cmd := exec.Command(command, submission.args...)
@@ -58,14 +58,14 @@ func runSparkSubmit(submission *submission) (bool, error) {
 		}
 		// The driver pod of the application already exists.
 		if strings.Contains(errorMsg, common.ErrorCodePodAlreadyExists) {
-			return false, fmt.Errorf("driver pod already exist")
+			return fmt.Errorf("driver pod already exist")
 		}
 		if errorMsg != "" {
-			return false, fmt.Errorf("failed to run spark-submit: %s", errorMsg)
+			return fmt.Errorf("failed to run spark-submit: %s", errorMsg)
 		}
-		return false, fmt.Errorf("failed to run spark-submit: %v", err)
+		return fmt.Errorf("failed to run spark-submit: %v", err)
 	}
-	return true, nil
+	return nil
 }
 
 // buildSparkSubmitArgs builds the arguments for spark-submit.
diff --git a/pkg/util/sparkapplication.go b/pkg/util/sparkapplication.go
index 29b8dab..a0aadd9 100644
--- a/pkg/util/sparkapplication.go
+++ b/pkg/util/sparkapplication.go
@@ -104,6 +104,27 @@ func ShouldRetry(app *v1beta2.SparkApplication) bool {
 	return false
 }
 
+func TimeUntilNextRetryDue(app *v1beta2.SparkApplication) (time.Duration, error) {
+	var retryInterval *int64
+	switch app.Status.AppState.State {
+	case v1beta2.ApplicationStateFailedSubmission:
+		retryInterval = app.Spec.RestartPolicy.OnSubmissionFailureRetryInterval
+	case v1beta2.ApplicationStateFailing:
+		retryInterval = app.Spec.RestartPolicy.OnFailureRetryInterval
+	}
+
+	attemptsDone := app.Status.SubmissionAttempts
+	lastAttemptTime := app.Status.LastSubmissionAttemptTime
+	if retryInterval == nil || lastAttemptTime.IsZero() || attemptsDone <= 0 {
+		return -1, fmt.Errorf("invalid retry interval (%v), last attempt time (%v) or attemptsDone (%v)", retryInterval, lastAttemptTime, attemptsDone)
+	}
+
+	// Retry wait time is attempts*RetryInterval to do a linear backoff.
+	interval := time.Duration(*retryInterval) * time.Second * time.Duration(attemptsDone)
+	currentTime := time.Now()
+	return interval - currentTime.Sub(lastAttemptTime.Time), nil
+}
+
 func GetLocalVolumes(app *v1beta2.SparkApplication) map[string]corev1.Volume {
 	volumes := make(map[string]corev1.Volume)
 	for _, volume := range app.Spec.Volumes {



Test patch:
diff --git a/charts/spark-operator-chart/tests/controller/deployment_test.yaml b/charts/spark-operator-chart/tests/controller/deployment_test.yaml
index c2513dd..67c475d 100644
--- a/charts/spark-operator-chart/tests/controller/deployment_test.yaml
+++ b/charts/spark-operator-chart/tests/controller/deployment_test.yaml
@@ -171,6 +171,19 @@ tests:
           path: spec.template.spec.containers[?(@.name=="spark-operator-controller")].args
           content: --ingress-url-format={{$appName}}.example.com/{{$appNamespace}}/{{$appName}}
 
+  - it: Should contain `--ingress-class-name` arg if `controller.uiIngress.enable` is set to `true` and `controller.uiIngress.ingressClassName` is set
+    set:
+      controller:
+        uiService:
+          enable: true
+        uiIngress:
+          enable: true
+          ingressClassName: nginx
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[?(@.name=="spark-operator-controller")].args
+          content: --ingress-class-name=nginx
+
   - it: Should contain `--enable-batch-scheduler` arg if `controller.batchScheduler.enable` is `true`
     set:
       controller:
diff --git a/test/e2e/bad_examples/fail-application.yaml b/test/e2e/bad_examples/fail-application.yaml
new file mode 100644
index 0000000..05991fd
--- /dev/null
+++ b/test/e2e/bad_examples/fail-application.yaml
@@ -0,0 +1,44 @@
+#
+# Copyright 2024 The Kubeflow authors.
+#
+# 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
+#
+#     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.
+
+apiVersion: sparkoperator.k8s.io/v1beta2
+kind: SparkApplication
+metadata:
+  name: fail-submission
+  namespace: default
+spec:
+  type: Scala
+  mode: cluster
+  image: spark:3.5.2
+  imagePullPolicy: IfNotPresent
+  mainClass: non-existent
+  mainApplicationFile: local:///non-existent.jar
+  sparkVersion: 3.5.2
+  restartPolicy:
+    type: OnFailure
+    onFailureRetries: 3
+    onFailureRetryInterval: 1
+  driver:
+    labels:
+      version: 3.5.2
+    cores: 1
+    memory: 512m
+    serviceAccount: spark-operator-spark
+  executor:
+    labels:
+      version: 3.5.2
+    instances: 1
+    cores: 1
+    memory: 512m
diff --git a/test/e2e/bad_examples/fail-submission.yaml b/test/e2e/bad_examples/fail-submission.yaml
new file mode 100644
index 0000000..827fca7
--- /dev/null
+++ b/test/e2e/bad_examples/fail-submission.yaml
@@ -0,0 +1,44 @@
+#
+# Copyright 2024 The Kubeflow authors.
+#
+# 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
+#
+#     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.
+
+apiVersion: sparkoperator.k8s.io/v1beta2
+kind: SparkApplication
+metadata:
+  name: fail-submission
+  namespace: default
+spec:
+  type: Scala
+  mode: cluster
+  image: spark:3.5.2
+  imagePullPolicy: IfNotPresent
+  mainClass: dummy
+  mainApplicationFile: local:///dummy.jar
+  sparkVersion: 3.5.2
+  restartPolicy:
+    type: OnFailure
+    onSubmissionFailureRetries: 3
+    onSubmissionFailureRetryInterval: 1
+  driver:
+    labels:
+      version: 3.5.2
+    cores: 1
+    memory: 512m
+    serviceAccount: non-existent  # This is the important part that causes submission to fail.
+  executor:
+    labels:
+      version: 3.5.2
+    instances: 1
+    cores: 1
+    memory: 512m
diff --git a/test/e2e/sparkapplication_test.go b/test/e2e/sparkapplication_test.go
index 113326c..825129d 100644
--- a/test/e2e/sparkapplication_test.go
+++ b/test/e2e/sparkapplication_test.go
@@ -27,6 +27,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
+	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -236,6 +237,122 @@ var _ = Describe("Example SparkApplication", func() {
 		})
 	})
 
+	Context("fail-submission", func() {
+		ctx := context.Background()
+		path := filepath.Join("bad_examples", "fail-submission.yaml")
+		app := &v1beta2.SparkApplication{}
+
+		BeforeEach(func() {
+			By("Parsing SparkApplication from file")
+			file, err := os.Open(path)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(file).NotTo(BeNil())
+
+			decoder := yaml.NewYAMLOrJSONDecoder(file, 100)
+			Expect(decoder).NotTo(BeNil())
+			Expect(decoder.Decode(app)).NotTo(HaveOccurred())
+
+			By("Creating SparkApplication")
+			Expect(k8sClient.Create(ctx, app)).To(Succeed())
+		})
+
+		AfterEach(func() {
+			key := types.NamespacedName{Namespace: app.Namespace, Name: app.Name}
+			Expect(k8sClient.Get(ctx, key, app)).To(Succeed())
+
+			By("Deleting SparkApplication")
+			Expect(k8sClient.Delete(ctx, app)).To(Succeed())
+		})
+
+		It("Fails submission and retries until retries are exhausted", func() {
+			By("Waiting for SparkApplication to terminate")
+			key := types.NamespacedName{Namespace: app.Namespace, Name: app.Name}
+			apps, polling_err := collectSparkApplicationsUntilTermination(ctx, key)
+			Expect(polling_err).To(HaveOccurred())
+
+			By("Should eventually fail")
+			finalApp := apps[len(apps)-1]
+			Expect(finalApp.Status.AppState.State).To(Equal(v1beta2.ApplicationStateFailed))
+			Expect(finalApp.Status.AppState.ErrorMessage).To(ContainSubstring("failed to run spark-submit"))
+			Expect(finalApp.Status.SubmissionAttempts).To(Equal(*app.Spec.RestartPolicy.OnSubmissionFailureRetries + 1))
+
+			By("Only valid statuses appear in other apps")
+			validStatuses := []v1beta2.ApplicationStateType{
+				v1beta2.ApplicationStateNew,
+				v1beta2.ApplicationStateFailedSubmission,
+			}
+			for _, app := range apps[:len(apps)-1] {
+				Expect(validStatuses).To(ContainElement(app.Status.AppState.State))
+			}
+
+			By("Checking driver does not exist")
+			driverPodName := util.GetDriverPodName(app)
+			driverPodKey := types.NamespacedName{Namespace: app.Namespace, Name: driverPodName}
+			err := k8sClient.Get(ctx, driverPodKey, &corev1.Pod{})
+			Expect(errors.IsNotFound(err)).To(BeTrue())
+		})
+	})
+
+	Context("application-fails", func() {
+		ctx := context.Background()
+		path := filepath.Join("bad_examples", "fail-application.yaml")
+		app := &v1beta2.SparkApplication{}
+
+		BeforeEach(func() {
+			By("Parsing SparkApplication from file")
+			file, err := os.Open(path)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(file).NotTo(BeNil())
+
+			decoder := yaml.NewYAMLOrJSONDecoder(file, 100)
+			Expect(decoder).NotTo(BeNil())
+			Expect(decoder.Decode(app)).NotTo(HaveOccurred())
+
+			By("Creating SparkApplication")
+			Expect(k8sClient.Create(ctx, app)).To(Succeed())
+		})
+
+		AfterEach(func() {
+			key := types.NamespacedName{Namespace: app.Namespace, Name: app.Name}
+			Expect(k8sClient.Get(ctx, key, app)).To(Succeed())
+
+			By("Deleting SparkApplication")
+			Expect(k8sClient.Delete(ctx, app)).To(Succeed())
+		})
+
+		It("Application fails and retries until retries are exhausted", func() {
+			By("Waiting for SparkApplication to terminate")
+			key := types.NamespacedName{Namespace: app.Namespace, Name: app.Name}
+			apps, polling_err := collectSparkApplicationsUntilTermination(ctx, key)
+			Expect(polling_err).To(HaveOccurred())
+
+			By("Should eventually fail")
+			final_app := apps[len(apps)-1]
+			Expect(final_app.Status.AppState.State).To(Equal(v1beta2.ApplicationStateFailed))
+			Expect(final_app.Status.AppState.ErrorMessage).To(ContainSubstring("driver container failed"))
+			Expect(final_app.Status.ExecutionAttempts).To(Equal(*app.Spec.RestartPolicy.OnFailureRetries + 1))
+
+			By("Only valid statuses appear in other apps")
+			validStatuses := []v1beta2.ApplicationStateType{
+				v1beta2.ApplicationStateNew,
+				v1beta2.ApplicationStateSubmitted,
+				v1beta2.ApplicationStateRunning,
+				v1beta2.ApplicationStateFailing,
+				v1beta2.ApplicationStatePendingRerun,
+			}
+			for _, app := range apps[:len(apps)-1] {
+				Expect(validStatuses).To(ContainElement(app.Status.AppState.State))
+			}
+
+			By("Checking out driver logs")
+			driverPodName := util.GetDriverPodName(app)
+			bytes, err := clientset.CoreV1().Pods(app.Namespace).GetLogs(driverPodName, &corev1.PodLogOptions{}).Do(ctx).Raw()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(bytes).NotTo(BeEmpty())
+			Expect(strings.Contains(string(bytes), "NoSuchFileException")).To(BeTrue())
+		})
+	})
+
 	Context("spark-pi-python", func() {
 		ctx := context.Background()
 		path := filepath.Join("..", "..", "examples", "spark-pi-python.yaml")
diff --git a/test/e2e/suit_test.go b/test/e2e/suit_test.go
index 9216708..e9c534d 100644
--- a/test/e2e/suit_test.go
+++ b/test/e2e/suit_test.go
@@ -271,3 +271,26 @@ func waitForSparkApplicationCompleted(ctx context.Context, key types.NamespacedN
 	})
 	return err
 }
+
+func collectSparkApplicationsUntilTermination(ctx context.Context, key types.NamespacedName) ([]v1beta2.SparkApplication, error) {
+	cancelCtx, cancelFunc := context.WithTimeout(ctx, WaitTimeout)
+	defer cancelFunc()
+
+	apps := []v1beta2.SparkApplication{}
+
+	err := wait.PollUntilContextCancel(cancelCtx, PollInterval, true, func(ctx context.Context) (bool, error) {
+		app := v1beta2.SparkApplication{}
+		if err := k8sClient.Get(ctx, key, &app); err != nil {
+			return false, err
+		}
+		apps = append(apps, app)
+		switch app.Status.AppState.State {
+		case v1beta2.ApplicationStateFailed:
+			return true, errors.New(app.Status.AppState.ErrorMessage)
+		case v1beta2.ApplicationStateCompleted:
+			return true, nil
+		}
+		return false, nil
+	})
+	return apps, err
+}