Skip to content

Commit f83312e

Browse files
committed
[SPARK-60052] Suspend and requeue a running SparkApplication on Kueue eviction
### What changes were proposed in this pull request? This PR aims to suspend a running `SparkApplication` whose Kueue `Workload` is evicted or deactivated, and to queue it again with a new attempt, like a running `SparkCluster`. - `AppSuspendStep` moves such an application to `Suspended` and releases its driver and executors. Then it deletes the `Workload` after the requeue backoff, or keeps a deactivated one until it is reactivated. - The new attempt starts with `ApplicationStatus#resume`, so it does not count as a restart. - A `PodsReadyTimeout` is handled like any other eviction, and `spec.suspend` takes precedence. - `AppKueueEvictionStep` and the `KueueEvictionIgnored` event are removed. ### Why are the changes needed? Currently, the driver and executors keep running after an eviction. So preemption, cohort quota reclamation and `waitForPodsReady` do not work, and a deactivated `Workload` over-commits the quota. ### Does this PR introduce _any_ user-facing change? No for the released versions, because the Kueue integration is not released yet in [1.0.0](https://github.com/apache/spark-kubernetes-operator/releases/tag/1.0.0) (2026-07-23). ### How was this patch tested? Pass the CIs with the newly added test cases. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 5.5 Closes #952 from dongjoon-hyun/SPARK-60052. Authored-by: Dongjoon Hyun <dongjoon@apache.org> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>
1 parent cb87348 commit f83312e

18 files changed

Lines changed: 700 additions & 368 deletions

File tree

‎docs/architecture.md‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -138,8 +138,8 @@ stateDiagram-v2
138138
RunningWithPartialCapacity --> Succeeded
139139
RunningWithPartialCapacity --> Failed
140140
141-
RunningHealthy --> Suspended : spec.suspend=true
142-
Suspended --> Submitted : spec.suspend=false, new attempt
141+
RunningHealthy --> Suspended : spec.suspend=true or Kueue eviction
142+
Suspended --> Submitted : spec.suspend=false or after Kueue eviction, new attempt
143143
144144
state Failures {
145145
SchedulingFailure
@@ -171,7 +171,10 @@ stateDiagram-v2
171171
* Likewise, an application moves to `Suspended` from any of the states from `DriverRequested` to
172172
`RunningWithBelowThresholdExecutors` when [`spec.suspend`](spark_custom_resources.md#suspend) is
173173
set to `true`. Its driver and executors are released, and setting it back to `false` starts a
174-
new attempt from `Submitted`, which does not count as a restart.
174+
new attempt from `Submitted`, which does not count as a restart. An application queued by
175+
[Kueue](spark_custom_resources.md#kueue) is suspended as well when Kueue evicts its `Workload`,
176+
and starts such a new attempt once its driver and executors are released and the requeue
177+
backoff of its `Workload` elapsed, or once its `Workload` is reactivated if it was deactivated.
175178
* User may configure the app CR to time-out after given threshold of time if it cannot reach healthy
176179
state after given threshold. The timeout can be configured for different lifecycle stages,
177180
when driver starting, when driver becoming ready, and when requesting executor pods.

‎docs/config_properties.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@
5353
| spark.kubernetes.operator.reconciler.retry.initialIntervalSeconds | Integer | 5 | false | Initial interval(in seconds) of retries on unhandled controller errors. |
5454
| spark.kubernetes.operator.reconciler.retry.intervalMultiplier | Double | 1.5 | false | Interval multiplier of retries on unhandled controller errors. Setting this to 1 for linear retry. |
5555
| spark.kubernetes.operator.reconciler.retry.maxIntervalSeconds | Integer | -1 | false | Max interval(in seconds) of retries on unhandled controller errors. Set to non-positive for unlimited. |
56-
| spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds | Long | 1800 | true | Requeue interval (in seconds) at which a resource held by spec.suspend is reconciled, so that its SuspendHeld event is republished. A SparkApplication or SparkCluster suspended while running publishes no such event, since its Suspended status says so, and uses this interval only to look at its released resources again. The hold ends only when a user clears spec.suspend, which arrives as a watch event and reconciles right away, so nothing waits for this interval. It is deliberately much coarser than spark.kubernetes.operator.reconciler.intervalSeconds, since a suspended resource has nothing to observe while each republish costs a read and a write on the API server. Keep it below the '--event-ttl' of the API server (one hour by default), or the event expires between repeats and the hold stops being visible, and keep 'spark.kubernetes.operator.events.minIntervalSeconds' within the rest of that TTL, that is at most the TTL minus this interval, or a repeat is dropped and the next one lands after the event expired. |
56+
| spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds | Long | 1800 | true | Requeue interval (in seconds) at which a resource held by spec.suspend is reconciled, so that its SuspendHeld event is republished. A SparkApplication or SparkCluster suspended while running publishes no such event, since its Suspended status says so, and uses this interval only to look at its released resources again. The hold ends only when a user clears spec.suspend, which arrives as a watch event and reconciles right away, so nothing waits for this interval. It also paces a SparkApplication or SparkCluster which is suspended by a Kueue eviction and whose Workload is deactivated. That hold ends when the Workload is reactivated, which arrives as a watch event as well. It is deliberately much coarser than spark.kubernetes.operator.reconciler.intervalSeconds, since a suspended resource has nothing to observe while each republish costs a read and a write on the API server. Keep it below the '--event-ttl' of the API server (one hour by default), or the event expires between repeats and the hold stops being visible, and keep 'spark.kubernetes.operator.events.minIntervalSeconds' within the rest of that TTL, that is at most the TTL minus this interval, or a repeat is dropped and the next one lands after the event expired. |
5757
| spark.kubernetes.operator.reconciler.terminationTimeoutSeconds | Integer | 30 | false | Grace period for operator shutdown before reconciliation threads are killed. |
5858
| spark.kubernetes.operator.reconciler.trimStateTransitionHistoryEnabled | Boolean | true | true | When enabled, operator would trim state transition history when a new attempt starts, keeping previous attempt summary only. It also drops the history of a suspended SparkCluster when it is resumed. |
5959
| spark.kubernetes.operator.watchedNamespaces | String | default | true | Comma-separated list of namespaces that the operator would be watching for Spark resources. If set to '*', operator would watch all namespaces. |

‎docs/configuration.md‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ replacing its `message`. A repeat carrying the same `message` may be paced by
102102
The initial `Submitted` status of a new resource is not persisted to the API server on its own, so
103103
no event is published for it. A `Submitted` event is published only when a `SparkApplication` or a
104104
`SparkCluster` moves from `Suspended` back to `Submitted`, after `spec.suspend` is set back to
105-
`false`, or after the eviction of a `SparkCluster` by Kueue.
105+
`false`, or after its eviction by Kueue.
106106

107107
In addition, the operator publishes the following `Warning` events.
108108

@@ -115,9 +115,8 @@ In addition, the operator publishes the following `Warning` events.
115115
| `ClusterRequestFailed` | Applying the Services, StatefulSets, NetworkPolicy, HorizontalPodAutoscaler or PodDisruptionBudget of a `SparkCluster` fails in a way which may yet succeed, such as a throttled request or an internal server error from an admission webhook, so it is retried with the default interval instead of failing the cluster. Transport-level errors are skipped. |
116116
| `SuspendReleaseFailed` | Deleting the driver pod (`SparkApplication`), the master and worker StatefulSets or the HorizontalPodAutoscaler or PodDisruptionBudget of the workers (`SparkCluster`), or the Kueue `Workload` of a resource suspended while running fails, or its pods cannot be listed. A failure which may clear on its own, such as a timeout, is skipped. Every failure is retried with the default interval. |
117117
| `KueueResourceFlavorReadFailed` | Reading the `ResourceFlavor`s which Kueue assigned to an admitted `Workload` fails, so the node selector and the tolerations of the flavors cannot be applied. Transport-level errors are skipped and retried every 5 seconds, while a persistent failure is retried with the default interval. |
118-
| `KueuePodsReadyUpdateFailed` | Recording the `PodsReady` condition on the Kueue `Workload` of a running `SparkApplication` or `SparkCluster` fails, e.g. without the permission for the `workloads/status` subresource, so Kueue may evict it by a `PodsReadyTimeout`, which releases a running `SparkCluster` and queues it again even if its master and workers are ready. Transport-level errors and conflicts are skipped. A `SparkCluster` retries them every 5 seconds and a persistent failure with the default interval, while a `SparkApplication` retries any failure with its next reconciliation, so that the recording does not hold back the observation of its driver. See [Kueue](spark_custom_resources.md#kueue). |
118+
| `KueuePodsReadyUpdateFailed` | Recording the `PodsReady` condition on the Kueue `Workload` of a running `SparkApplication` or `SparkCluster` fails, e.g. without the permission for the `workloads/status` subresource, so Kueue may evict it by a `PodsReadyTimeout`, which releases it and queues it again, a `SparkApplication` with a new attempt, even if its pods are ready. Transport-level errors and conflicts are skipped. A `SparkCluster` retries them every 5 seconds and a persistent failure with the default interval, while a `SparkApplication` retries any failure with its next reconciliation, so that the recording does not hold back the observation of its driver. See [Kueue](spark_custom_resources.md#kueue). |
119119
| `KueueDisabled` | A `SparkApplication` or `SparkCluster` labeled with `kueue.x-k8s.io/queue-name` is not queued, since `spark.kubernetes.operator.kueue.enabled` is disabled, so the driver (or master and worker) is requested without the Kueue admission. |
120-
| `KueueEvictionIgnored` | Kueue evicted or deactivated the `Workload` of a `SparkApplication` whose driver is requested and whose attempt has neither stopped nor been suspended. The operator does not act on it, so the driver and executors keep running. An evicted `Workload` keeps holding its quota, while a deactivated one no longer counts against it. It is republished on every reconciliation while it lasts, subject to [`minIntervalSeconds`](#event-frequency). See [Kueue](spark_custom_resources.md#kueue). |
121120

122121
For a resource held by [`spec.suspend`](spark_custom_resources.md#suspend) or queued by
123122
[Kueue](spark_custom_resources.md#kueue), the operator also publishes the following `Normal`

‎docs/spark_custom_resources.md‎

Lines changed: 20 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -829,18 +829,26 @@ spec:
829829
condition cannot be recorded, is admitted and evicted again and again until it is fixed,
830830
suspended or deleted. With `blockAdmission`, it holds back every other workload from each
831831
admission until its `Workload` is deleted after the backoff.
832-
* Preemption is not honored yet after the driver of a `SparkApplication` is requested, and neither
833-
is a `PodsReadyTimeout`, e.g. of an application which runs with fewer executors than
834-
`spark.executor.instances`, or a deactivation. The operator checks the admission only before it
835-
creates the resources, so after a later eviction the driver and executors keep running, and
836-
until the attempt stops or is suspended, the `KueueEvictionIgnored`
837-
[event](configuration.md#kubernetes-events) is published instead by default. An evicted
838-
`Workload` keeps its Kueue quota until the driver and executors are released, so a workload
839-
which preempts the application waits until then, while a deactivated one no longer counts
840-
against it. Set `spec.suspend` to `true`, as described in [Suspend](#suspend), or delete the
841-
application to release its quota earlier. With `blockAdmission`, an
842-
application whose pods are not all ready holds back every other workload until they are ready
843-
or its quota is released as described above.
832+
* A `SparkApplication` whose driver is requested, i.e. from `DriverRequested` to
833+
`RunningWithBelowThresholdExecutors`, is released like a running `SparkCluster` above when Kueue
834+
evicts its `Workload`, e.g. to preempt it or by a `PodsReadyTimeout`, or deactivates it. The
835+
application enters `Suspended` with a message giving the reason of the eviction, and the
836+
operator deletes its driver pod, which deletes its executor pods. Like for a cluster, the
837+
`Workload` is deleted once the pods are gone and its requeue backoff elapsed, or kept until it is
838+
reactivated if it was deactivated, and then the application starts a new attempt from
839+
`Submitted`, which is queued again with a new `Workload`. Like a resumed one, the new attempt
840+
runs the application again from scratch and does not count against the restart limits, so it
841+
starts even with `restartPolicy: Never`. An attempt whose driver has completed or failed by then
842+
ends as usual instead, and `spec.suspend` takes precedence, so an application which is suspended
843+
by it as well stays `Suspended` until it is resumed rather than being queued again.
844+
* Unlike without Kueue, a `SparkApplication` whose driver and `spark.executor.instances` executors
845+
are not all ready within the `waitForPodsReady` timeout, e.g. since some executors cannot be
846+
scheduled, does not keep running with fewer executors, even if its `applicationTolerations`
847+
allow it, e.g. in `RunningWithPartialCapacity`. Its `Workload` requested the quota for all of
848+
them, so the `PodsReadyTimeout` releases the application as above, and it runs again from
849+
scratch with a new attempt, which waits for the quota of all of them again. Like such a cluster,
850+
an application whose executors never get ready is admitted and evicted again and again, with a
851+
new attempt each time, until it is fixed, suspended or deleted.
844852
* To limit the execution time of a `SparkApplication`, the Spark native `spark.driver.timeout` is
845853
recommended instead of the `kueue.x-k8s.io/max-exec-time-seconds` label, which is not copied to
846854
the `Workload`. It requires `spark.plugins=org.apache.spark.deploy.DriverTimeoutPlugin`, as in

‎spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -246,6 +246,22 @@ public final class Constants {
246246
"Application is resumed as spec.suspend is set to false, so a new attempt starts from "
247247
+ "scratch.";
248248

249+
/**
250+
* Message indicating that the application is evicted by Kueue after its driver was requested,
251+
* which is followed by the reason and the message of the eviction. The operator tells such an
252+
* application apart from one suspended by spec.suspend by this prefix of its persisted state
253+
* message, so a changed wording no longer matches the applications suspended before.
254+
*/
255+
public static final String APP_EVICTED_MESSAGE =
256+
"Application is evicted by Kueue, so its driver and executors are being released before its "
257+
+ "Workload. It is queued again with a new attempt then, or once the Workload is "
258+
+ "reactivated if it was deactivated.";
259+
260+
/** Message indicating that the application evicted by Kueue is submitted again. */
261+
public static final String APP_REQUEUED_MESSAGE =
262+
"Application is submitted again after its eviction by Kueue, so a new attempt starts from "
263+
+ "scratch.";
264+
249265
// Spark Cluster Messages
250266
/** Message indicating a failure to request the Spark cluster from the scheduler backend. */
251267
public static final String CLUSTER_SCHEDULE_FAILURE_MESSAGE =

‎spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -112,10 +112,10 @@ public enum ApplicationStateSummary implements BaseStateSummary {
112112
TerminatedWithoutReleaseResources,
113113

114114
/**
115-
* The application is suspended by spec.suspend after its driver was requested, so its driver
116-
* and executors are released until it is resumed with a new attempt from Submitted. It is
117-
* declared last, after the terminated states, so that the ordinal ranges of {@link
118-
* #isStarting()} and {@link #isStopping()} exclude it.
115+
* The application is suspended by spec.suspend, or by the eviction of its Kueue Workload, after
116+
* its driver was requested, so its driver and executors are released until it is resumed with a
117+
* new attempt from Submitted. It is declared last, after the terminated states, so that the
118+
* ordinal ranges of {@link #isStarting()} and {@link #isStopping()} exclude it.
119119
*/
120120
Suspended;
121121

‎spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java‎

Lines changed: 13 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -268,13 +268,15 @@ public final class SparkOperatorConf {
268268
* suspended while running publishes no such event, since its Suspended status says so, and uses
269269
* this interval only to look at its released resources again. The hold ends only when a user
270270
* clears {@code spec.suspend}, which arrives as a watch event and reconciles right away, so
271-
* nothing waits for this interval. It is deliberately much coarser than {@link
272-
* #RECONCILER_INTERVAL_SECONDS}, since a suspended resource has nothing to observe while each
273-
* republish costs a read and a write on the API server. Keep it below the {@code --event-ttl} of
274-
* the API server (one hour by default), or the event expires between repeats and the hold stops
275-
* being visible, and keep {@link #KUBERNETES_EVENTS_MIN_INTERVAL_SECONDS} within the rest of
276-
* that TTL, that is at most the TTL minus this interval, or a repeat is dropped and the next
277-
* one lands after the event expired.
271+
* nothing waits for this interval. It also paces a SparkApplication or SparkCluster which is
272+
* suspended by a Kueue eviction and whose Workload is deactivated. That hold ends when the
273+
* Workload is reactivated, which arrives as a watch event as well. It is deliberately much
274+
* coarser than {@link #RECONCILER_INTERVAL_SECONDS}, since a suspended resource has nothing to
275+
* observe while each republish costs a read and a write on the API server. Keep it below the
276+
* {@code --event-ttl} of the API server (one hour by default), or the event expires between
277+
* repeats and the hold stops being visible, and keep {@link
278+
* #KUBERNETES_EVENTS_MIN_INTERVAL_SECONDS} within the rest of that TTL, that is at most the TTL
279+
* minus this interval, or a repeat is dropped and the next one lands after the event expired.
278280
*/
279281
public static final ConfigOption<Long> SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS =
280282
ConfigOption.<Long>builder()
@@ -288,7 +290,10 @@ public final class SparkOperatorConf {
288290
+ "says so, and uses this interval only to look at its released resources "
289291
+ "again. The hold ends only "
290292
+ "when a user clears spec.suspend, which arrives as a watch event and "
291-
+ "reconciles right away, so nothing waits for this interval. It is "
293+
+ "reconciles right away, so nothing waits for this interval. It also paces a "
294+
+ "SparkApplication or SparkCluster which is suspended by a Kueue eviction and "
295+
+ "whose Workload is deactivated. That hold ends when the Workload is "
296+
+ "reactivated, which arrives as a watch event as well. It is "
292297
+ "deliberately much coarser than "
293298
+ "spark.kubernetes.operator.reconciler.intervalSeconds, since a suspended "
294299
+ "resource has nothing to observe while each republish costs a read and a "

‎spark-operator/src/main/java/org/apache/spark/k8s/operator/kueue/KueueWorkloadUtils.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -923,7 +923,7 @@ public static boolean finishWorkload(
923923
* timeout of Kueue does not evict it, and with `blockAdmission`, other workloads are admitted.
924924
* Unlike the built-in Kueue integrations, the condition is not set back to `False` when a pod is
925925
* lost later, e.g. an executor which Spark replaces, since an eviction would restart a running
926-
* cluster with its applications, and is not acted on for a running application.
926+
* cluster with its applications, or the attempt of a running application from scratch.
927927
*
928928
* <p>The informer cache answers whether the condition is still missing, so that a resource which
929929
* reports it already, or whose pods are not ready yet, costs no request. Like {@link

‎spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,6 @@
5555
import org.apache.spark.k8s.operator.reconciler.observers.AppDriverTimeoutObserver;
5656
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppCleanUpStep;
5757
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppInitStep;
58-
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppKueueEvictionStep;
5958
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppReconcileStep;
6059
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppResourceObserveStep;
6160
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppRunningStep;
@@ -207,7 +206,6 @@ protected List<AppReconcileStep> getReconcileSteps(final SparkApplication app) {
207206
switch (app.getStatus().getCurrentState().getCurrentStateSummary()) {
208207
case Submitted, ScheduledToRestart -> steps.add(new AppInitStep());
209208
case DriverRequested, DriverStarted -> {
210-
steps.add(new AppKueueEvictionStep());
211209
steps.add(
212210
new AppResourceObserveStep(
213211
List.of(new AppDriverStartObserver(), new AppDriverReadyObserver())));
@@ -220,7 +218,6 @@ protected List<AppReconcileStep> getReconcileSteps(final SparkApplication app) {
220218
RunningHealthy,
221219
RunningWithPartialCapacity,
222220
RunningWithBelowThresholdExecutors -> {
223-
steps.add(new AppKueueEvictionStep());
224221
steps.add(new AppRunningStep());
225222
steps.add(new AppResourceObserveStep(List.of(new AppDriverRunningObserver())));
226223
steps.add(new AppResourceObserveStep(List.of(new AppDriverTimeoutObserver())));

0 commit comments

Comments
 (0)