This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new c5bc180 [SPARK-57492] Use K8s Garbage Collection to delete executors
only when restart is disabled
c5bc180 is described below
commit c5bc180122f7852353f9174993fca7eaf75b137e
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Jun 16 09:09:55 2026 -0700
[SPARK-57492] Use K8s Garbage Collection to delete executors only when
restart is disabled
### What changes were proposed in this pull request?
This PR applies the `spark.kubernetes.executor.deleteOnTermination=false`
override from
SPARK-55352 only when the application does not restart, i.e.
`RestartPolicy` is `Never` or
unset, in addition to the existing static-allocation condition.
### Why are the changes needed?
When `RestartPolicy=Always` (like `spark-thrift-server.yaml`), the driver
pod can be recreated while old executor pods linger, so they accumulate instead
of being garbage collected. Such applications must keep the default behavior
which delete executors on termination explicitly.
https://github.com/apache/spark-kubernetes-operator/blob/e875263375e5cd5bac7e160d8d7b913b9b9c28e6/examples/spark-thrift-server.yaml#L34
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Pass the CIs with the updated `useGarbageCollectionToDeleteExecutors` unit
test.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #708 from dongjoon-hyun/SPARK-57492.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../k8s/operator/SparkAppSubmissionWorker.java | 11 +++++++----
.../k8s/operator/SparkAppSubmissionWorkerTest.java | 22 +++++++++++++++++-----
2 files changed, 24 insertions(+), 9 deletions(-)
diff --git
a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
index 7bf8361..708f36d 100644
---
a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
+++
b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
@@ -40,7 +40,8 @@ import org.apache.spark.deploy.k8s.submit.RMainAppResource;
import org.apache.spark.k8s.operator.spec.ApplicationSpec;
import org.apache.spark.k8s.operator.spec.ConfigMapSpec;
import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
-import org.apache.spark.k8s.operator.spec.ResourceRetainPolicy;
+import org.apache.spark.k8s.operator.spec.RestartConfig;
+import org.apache.spark.k8s.operator.spec.RestartPolicy;
import org.apache.spark.k8s.operator.spec.RuntimeVersions;
import org.apache.spark.k8s.operator.utils.ModelUtils;
import org.apache.spark.k8s.operator.utils.StringUtils;
@@ -169,10 +170,12 @@ public class SparkAppSubmissionWorker {
effectiveSparkConf.setIfMissing("spark.io.encryption.enabled", "true");
effectiveSparkConf.setIfMissing("spark.kubernetes.driver.annotateExitException",
"true");
effectiveSparkConf.setIfMissing("spark.kubernetes.executor.useDriverPodIP",
"true");
- // In case of static allocation, use K8s Garbage Collection instead of
explicit API invocations
+ // In case of static allocation without restart, use K8s Garbage
Collection instead of
+ // explicit API invocations. If the application is configured to restart,
the driver pod may be
+ // recreated while executor pods linger, so executors must be deleted on
termination explicitly.
+ RestartConfig restartConfig =
applicationSpec.getApplicationTolerations().getRestartConfig();
if
(!"true".equalsIgnoreCase(effectiveSparkConf.get("spark.dynamicAllocation.enabled",
"false"))
- &&
applicationSpec.getApplicationTolerations().getResourceRetainPolicy() !=
- ResourceRetainPolicy.Always) {
+ && (restartConfig == null || RestartPolicy.Never ==
restartConfig.getRestartPolicy())) {
effectiveSparkConf.setIfMissing("spark.kubernetes.executor.deleteOnTermination",
"false");
}
return SparkAppDriverConf.create(
diff --git
a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorkerTest.java
b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorkerTest.java
index a6c89bb..bffe9ed 100644
---
a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorkerTest.java
+++
b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorkerTest.java
@@ -44,7 +44,8 @@ import
org.apache.spark.deploy.k8s.submit.PythonMainAppResource;
import org.apache.spark.deploy.k8s.submit.RMainAppResource;
import org.apache.spark.k8s.operator.spec.ApplicationSpec;
import org.apache.spark.k8s.operator.spec.ApplicationTolerations;
-import org.apache.spark.k8s.operator.spec.ResourceRetainPolicy;
+import org.apache.spark.k8s.operator.spec.RestartConfig;
+import org.apache.spark.k8s.operator.spec.RestartPolicy;
import org.apache.spark.k8s.operator.spec.RuntimeVersions;
import org.apache.spark.k8s.operator.status.ApplicationAttemptInfo;
import org.apache.spark.k8s.operator.status.ApplicationAttemptSummary;
@@ -308,17 +309,28 @@ class SparkAppSubmissionWorkerTest {
when(mockApp.getMetadata()).thenReturn(appMeta);
SparkAppSubmissionWorker submissionWorker = new SparkAppSubmissionWorker();
-
when(mockTolerations.getResourceRetainPolicy()).thenReturn(ResourceRetainPolicy.Always);
+ // Restart enabled (Always): driver may be recreated, so executors are
deleted explicitly.
+ when(mockTolerations.getRestartConfig())
+
.thenReturn(RestartConfig.builder().restartPolicy(RestartPolicy.Always).build());
SparkAppDriverConf conf1 = submissionWorker.buildDriverConf(mockApp,
Map.of());
assertEquals("true",
conf1.get("spark.kubernetes.executor.deleteOnTermination", "true"));
-
when(mockTolerations.getResourceRetainPolicy()).thenReturn(ResourceRetainPolicy.Never);
+ // Restart disabled (Never): rely on K8s Garbage Collection.
+ when(mockTolerations.getRestartConfig())
+
.thenReturn(RestartConfig.builder().restartPolicy(RestartPolicy.Never).build());
SparkAppDriverConf conf2 = submissionWorker.buildDriverConf(mockApp,
Map.of());
assertEquals("false",
conf2.get("spark.kubernetes.executor.deleteOnTermination", "true"));
-
when(mockTolerations.getResourceRetainPolicy()).thenReturn(ResourceRetainPolicy.OnFailure);
+ // Restart enabled (OnFailure): driver may be recreated, so executors are
deleted explicitly.
+ when(mockTolerations.getRestartConfig())
+
.thenReturn(RestartConfig.builder().restartPolicy(RestartPolicy.OnFailure).build());
SparkAppDriverConf conf3 = submissionWorker.buildDriverConf(mockApp,
Map.of());
- assertEquals("false",
conf3.get("spark.kubernetes.executor.deleteOnTermination", "true"));
+ assertEquals("true",
conf3.get("spark.kubernetes.executor.deleteOnTermination", "true"));
+
+ // RestartConfig unset: rely on K8s Garbage Collection.
+ when(mockTolerations.getRestartConfig()).thenReturn(null);
+ SparkAppDriverConf conf4 = submissionWorker.buildDriverConf(mockApp,
Map.of());
+ assertEquals("false",
conf4.get("spark.kubernetes.executor.deleteOnTermination", "true"));
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
