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 ebdd9281 [SPARK-60025] Support `spec.suspend` for a running
`SparkApplication`
ebdd9281 is described below
commit ebdd928193aa2d1fb21c6105b007ed4b7972e3b9
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Oct 7 08:28:06 2026 -0700
[SPARK-60025] Support `spec.suspend` for a running `SparkApplication`
### What changes were proposed in this pull request?
This PR aims to support `spec.suspend` for a running `SparkApplication`.
- Add `AppSuspendStep`, which moves a running application to `Suspended`
and releases its driver, executors, and Kueue `Workload`. When `spec.suspend`
is set back to `false`, it starts a new attempt from `Submitted` with
`ApplicationStatus#resume`.
- Route `Suspended` to `AppSuspendStep` in `SparkAppReconciler`, and skip
the restart backoff for a resumed attempt in `AppInitStep`.
- Update the documentation and the e2e tests.
### Why are the changes needed?
Currently, `spec.suspend` takes effect only before the driver is requested.
So a running attempt keeps its resources, including its Kueue quota, until it
ends. A running `SparkCluster` can already be suspended.
### Does this PR introduce _any_ user-facing change?
Yes, within the unreleased `spec.suspend`. Setting `spec.suspend` to `true`
now stops the running attempt of a `SparkApplication`, and setting it back to
`false` starts a new attempt. There is no change compared to
[1.0.0](https://github.com/apache/spark-kubernetes-operator/releases/tag/1.0.0)
(2026-07-23), which has no `spec.suspend`.
### 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 #948 from dongjoon-hyun/SPARK-60025.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../kueue-validating-admission-policy.yaml | 8 +-
docs/architecture.md | 7 +
docs/config_properties.md | 2 +-
docs/configuration.md | 10 +-
docs/spark_custom_resources.md | 80 ++-
.../org/apache/spark/k8s/operator/Constants.java | 16 +
.../k8s/operator/config/SparkOperatorConf.java | 15 +-
.../operator/reconciler/SparkAppReconciler.java | 4 +
.../reconciler/reconcilesteps/AppInitStep.java | 5 +-
.../reconcilesteps/AppKueueEvictionStep.java | 2 +-
.../reconcilesteps/AppReconcileStep.java | 2 +-
.../reconciler/reconcilesteps/AppSuspendStep.java | 280 ++++++++
.../reconciler/reconcilesteps/SuspendUtils.java | 5 +-
.../spark/k8s/operator/utils/EventUtils.java | 4 +-
.../reconciler/SparkAppReconcilerTest.java | 58 +-
.../reconciler/reconcilesteps/AppInitStepTest.java | 36 +
.../reconcilesteps/AppKueueEvictionStepTest.java | 1 +
.../reconcilesteps/AppSuspendStepTest.java | 729 +++++++++++++++++++++
tests/e2e/kueue/chainsaw-test.yaml | 70 ++
tests/e2e/suspend/chainsaw-test.yaml | 80 +++
tests/e2e/suspend/spark-example-running.yaml | 35 +
21 files changed, 1387 insertions(+), 62 deletions(-)
diff --git
a/build-tools/helm/spark-kubernetes-operator/templates/kueue-validating-admission-policy.yaml
b/build-tools/helm/spark-kubernetes-operator/templates/kueue-validating-admission-policy.yaml
index f89e3908..ef77bfe6 100644
---
a/build-tools/helm/spark-kubernetes-operator/templates/kueue-validating-admission-policy.yaml
+++
b/build-tools/helm/spark-kubernetes-operator/templates/kueue-validating-admission-policy.yaml
@@ -15,7 +15,7 @@
{{/*
Keep the queue-name label of a SparkApplication or SparkCluster from changing
once it started,
-until the application is scheduled to restart or the cluster is Suspended. See
the Kueue section
+until the application is scheduled to restart or the resource is Suspended.
See the Kueue section
of docs/spark_custom_resources.md.
*/}}
{{- if .Values.operatorRbac.kueue.enabled }}
@@ -60,14 +60,14 @@ spec:
? oldObject.status.currentState.currentStateSummary : ''
validations:
# The states in which ApplicationStateSummary or
ClusterStateSummary#isInitializing holds, i.e.
- # before an attempt starts, and ClusterStateSummary.Suspended. Keep them
in sync.
+ # before an attempt starts, and the Suspended state of both. Keep them in
sync.
- expression: >-
variables.oldQueue == variables.newQueue
|| variables.oldState in ['', 'Submitted', 'ScheduledToRestart',
'Suspended']
message: >-
The kueue.x-k8s.io/queue-name label can be changed only before the
resource starts, i.e.
- while it is Submitted or ScheduledToRestart, or while a SparkCluster
is Suspended, since
- the Kueue Workload admitted for it keeps its quota in that queue.
+ while it is Submitted or ScheduledToRestart, or while it is Suspended,
since the Kueue
+ Workload admitted for it keeps its quota in that queue.
reason: Forbidden
---
apiVersion: admissionregistration.k8s.io/v1
diff --git a/docs/architecture.md b/docs/architecture.md
index 6b611286..7d4b70a1 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -138,6 +138,9 @@ stateDiagram-v2
RunningWithPartialCapacity --> Succeeded
RunningWithPartialCapacity --> Failed
+ RunningHealthy --> Suspended : spec.suspend=true
+ Suspended --> Submitted : spec.suspend=false, new attempt
+
state Failures {
SchedulingFailure
DriverStartTimedOut
@@ -165,6 +168,10 @@ stateDiagram-v2
* Once the driver is requested, an application moves to `Succeeded`, `Failed`
or `DriverEvicted`
from any of the states above whenever the driver pod terminates or is
evicted. It also moves to
`Failed` if the driver pod is removed unexpectedly.
+* Likewise, an application moves to `Suspended` from any of the states from
`DriverRequested` to
+ `RunningWithBelowThresholdExecutors` when
[`spec.suspend`](spark_custom_resources.md#suspend) is
+ set to `true`. Its driver and executors are released, and setting it back to
`false` starts a
+ new attempt from `Submitted`, which does not count as a restart.
* User may configure the app CR to time-out after given threshold of time if
it cannot reach healthy
state after given threshold. The timeout can be configured for different
lifecycle stages,
when driver starting, when driver becoming ready, and when requesting
executor pods.
diff --git a/docs/config_properties.md b/docs/config_properties.md
index 60153a86..2a8285b9 100644
--- a/docs/config_properties.md
+++ b/docs/config_properties.md
@@ -53,7 +53,7 @@
| spark.kubernetes.operator.reconciler.retry.initialIntervalSeconds | Integer
| 5 | false | Initial interval(in seconds) of retries on unhandled controller
errors. |
| 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. |
| 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. |
- | 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
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 reconci [...]
+ | 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 wat [...]
| spark.kubernetes.operator.reconciler.terminationTimeoutSeconds | Integer |
30 | false | Grace period for operator shutdown before reconciliation threads
are killed. |
| 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. |
| 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. |
diff --git a/docs/configuration.md b/docs/configuration.md
index 591262c9..115d0815 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -100,9 +100,9 @@ replacing its `message`. A repeat carrying the same
`message` may be paced by
| `Normal` | All other states, e.g. `ScheduledToRestart`, `DriverRequested`,
`DriverStarted`, `DriverReady`, `InitializedBelowThresholdExecutors`,
`RunningHealthy`, `RunningWithPartialCapacity`, `Succeeded`, `ResourceReleased`
and `Suspended` |
The initial `Submitted` status of a new resource is not persisted to the API
server on its own, so
-no event is published for it. A `Submitted` event is published only when a
`SparkCluster` moves
-from `Suspended` back to `Submitted`, after `spec.suspend` is set back to
`false` or after its
-eviction by Kueue.
+no event is published for it. A `Submitted` event is published only when a
`SparkApplication` or a
+`SparkCluster` moves from `Suspended` back to `Submitted`, after
`spec.suspend` is set back to
+`false`, or after the eviction of a `SparkCluster` by Kueue.
In addition, the operator publishes the following `Warning` events.
@@ -113,11 +113,11 @@ In addition, the operator publishes the following
`Warning` events.
| `StatusUpdateFailed` | A status patch is rejected. Transport-level errors
are skipped. |
| `KueueAdmissionRequestFailed` | Creating, reading or deleting a stale Kueue
`Workload` fails, or the driver or master cannot be read before the admission
is requested. Transport-level errors are skipped and retried every 5 seconds,
while a persistent failure is retried with the default interval. |
| `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. |
-| `SuspendReleaseFailed` | Deleting the master and worker StatefulSets, the
HorizontalPodAutoscaler or PodDisruptionBudget of the workers, or the Kueue
`Workload` of a `SparkCluster` 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. |
+| `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. |
| `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. |
| `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 [...]
| `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. |
-| `KueueEvictionIgnored` | Kueue evicted or deactivated the `Workload` of a
`SparkApplication` whose driver is requested and whose attempt has not stopped.
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). |
+| `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_re [...]
For a resource held by [`spec.suspend`](spark_custom_resources.md#suspend) or
queued by
[Kueue](spark_custom_resources.md#kueue), the operator also publishes the
following `Normal`
diff --git a/docs/spark_custom_resources.md b/docs/spark_custom_resources.md
index ef0911ea..e5c19516 100644
--- a/docs/spark_custom_resources.md
+++ b/docs/spark_custom_resources.md
@@ -534,7 +534,7 @@ Both `SparkApplication` and `SparkCluster` support
`.spec.suspend`. When it is s
the driver pod or the master / worker StatefulSets are requested, the operator
keeps the resource in
its initializing state (`Submitted`, or `ScheduledToRestart` for an
application that is scheduled to
restart) and does not request them. Setting it back to `false` resumes the
regular lifecycle. A
-running `SparkCluster` is stopped instead, as described below.
+running `SparkApplication` or `SparkCluster` is stopped instead, as described
below.
`Submitted` here is the operator's in-memory view. For a valid resource
created with
`suspend: true`, the initial `Submitted` status is not persisted to the API
server, so
@@ -542,9 +542,10 @@ running `SparkCluster` is stopped instead, as described
below.
initialization resumes. Instead, the `SuspendHeld`
[event](configuration.md#kubernetes-events) is
published by default. Since the status is not there to fall back on, it is
republished every 30
minutes by default while the hold lasts, so that it outlives the event
retention of the API
-server. An application held later, in `ScheduledToRestart`, keeps the status
its previous attempt
-already wrote and gets the same event, since that status says that a restart
is due, not that the
-next attempt is withheld.
+server. An application held later, in `ScheduledToRestart`, or in `Submitted`
after it was resumed
+as described below, keeps the status which was already written and gets the
same event, since that
+status says that a restart is due or that the application is resumed, not that
the next attempt is
+withheld.
``` yaml
apiVersion: spark.apache.org/v1
@@ -559,10 +560,31 @@ spec:
sparkVersion: "4.2.0"
```
-* For a `SparkApplication`, `suspend` only takes effect before the driver
resources are requested.
- Setting it to `true` on a running application does not stop the current
attempt. If the
- application is configured to restart, the next attempt is held until
`suspend` is set back to
- `false`.
+* Setting it to `true` on a `SparkApplication` whose driver is requested, i.e.
from
+ `DriverRequested` to `RunningWithBelowThresholdExecutors`, stops its current
attempt. The
+ application enters `Suspended` first, and then the operator deletes its
driver pod, which deletes
+ its executor pods and the other resources which the driver owns, e.g. its
ConfigMaps and
+ Services, like at the end of an attempt, whatever its `resourceRetainPolicy`
is. The progress of
+ the attempt is lost. An attempt whose driver has completed or failed by then
ends as usual
+ instead, and if the application is configured to restart, its next attempt
is held as above. A
+ failure to delete the driver pod or the Kueue `Workload`, other than one
which may clear on its
+ own, publishes the `SuspendReleaseFailed`
[event](configuration.md#kubernetes-events) by default.
+* Setting it back to `false` starts a new attempt of the application from
`Submitted` once its
+ driver and executor pods are gone and its Kueue `Workload`, if any, is
released. A pod which is
+ still terminating `forceTerminationGracePeriodMillis` after its grace period
ended, e.g. on a lost
+ node, no longer holds the application, and a driver pod is force deleted
then. This is measured
+ per pod, so that the pods are waited for even if deleting them failed or
started late. The new
+ attempt runs the application again from scratch with its current spec. Like
a restart, it gets
+ the next attempt ID, and with
+ `spark.kubernetes.operator.reconciler.trimStateTransitionHistoryEnabled`,
the state transition
+ history of the suspended attempt moves to `previousAttemptSummary`. The
attempt ID is part of the
+ names of its driver resources unless `spark.app.id` is set, while the
resources which the spec
+ names, e.g. through `configMapSpecs`, keep their names. Unlike a restart, it
starts without the
+ restart backoff and does not count against the restart limits, so the
restart counters stay as
+ they are, and it starts even with `restartPolicy: Never`. Only the counter
of consecutive
+ scheduling failures starts over, since the driver of the suspended attempt
was requested.
+ `restartCounterResetMillis` and the timeouts of `applicationTimeoutConfig`
apply to the new
+ attempt on its own, so neither the run before the suspension nor the time
suspended counts.
* Setting it to `true` on a running `SparkCluster` (`RunningHealthy`) stops
it, since a cluster
runs until it is deleted and has no other way to give its resources back.
The cluster enters
`Suspended` first, and then the operator deletes its master and worker
StatefulSets, and the
@@ -584,11 +606,13 @@ spec:
a lost node, no longer holds the Kueue `Workload`, so that it does not hold
the quota forever. The
cluster still stays `Suspended` until such a pod is gone, since it keeps the
name of the master or
worker to create, and the message of its `Suspended` state names such pods.
-* An operator version without the `Suspended` state cannot read a
`SparkCluster` whose status
- has it, even after the cluster is resumed. A resumed `SparkCluster` keeps it
unless
- `spark.kubernetes.operator.reconciler.trimStateTransitionHistoryEnabled`
drops the history of
- its previous run. So delete such clusters, or resume the suspended ones with
that setting
- enabled, before downgrading the operator.
+* An operator version without the `Suspended` state cannot read a
`SparkApplication` or a
+ `SparkCluster` whose status has it, even after the resource is resumed. A
resumed
+ `SparkApplication` keeps it in its state transition history, or with
+ `spark.kubernetes.operator.reconciler.trimStateTransitionHistoryEnabled` in
that of
+ `previousAttemptSummary`, and a resumed `SparkCluster` keeps it unless that
setting drops the
+ history of its previous run. So delete such applications, and delete such
clusters or resume the
+ suspended ones with that setting enabled, before downgrading the operator.
* Deleting a suspended resource works as usual.
* This is the building block for external job queueing systems. See
[Kueue](#kueue) for the
built-in integration.
@@ -694,7 +718,9 @@ spec:
read, the resource is held and the read is retried.
* `spec.suspend` takes precedence. A suspended resource does not get a
`Workload`, and suspending
a queued resource deletes its `Workload` to release the quota. Suspending a
running
- `SparkCluster` deletes its `Workload` only after its master and worker pods
are gone, or are still
+ `SparkApplication` deletes its `Workload` only after its driver and executor
pods are gone, or
+ are still terminating `forceTerminationGracePeriodMillis` after their grace
period ended, and
+ suspending a running `SparkCluster` only after its master and worker pods
are gone, or are still
terminating 5 minutes after their grace period ended, since terminating pods
occupy the quota.
The `Workload` is deleted even if the `kueue.x-k8s.io/queue-name` label was
removed after the
admission. The resource is queued again when it is resumed with the
`kueue.x-k8s.io/queue-name`
@@ -705,10 +731,11 @@ spec:
`SparkApplication` or a `SparkCluster` once its status says that it started,
since the `Workload`
admitted for it keeps its quota in that queue. The label may change again
when a
`SparkApplication` is `ScheduledToRestart`, whose next attempt is queued
with a new `Workload`, or
- when a `SparkCluster` is `Suspended`. So the label of a terminated resource,
e.g. `Failed`, stays
- as it was. To move a running `SparkCluster` to another queue, set
`spec.suspend` to `true`, change
- the label once the cluster is `Suspended`, and then resume it, while a
running `SparkApplication`
- moves with its next attempt. Before a resource starts, the operator follows
the label as described
+ when a `SparkApplication` or a `SparkCluster` is `Suspended`. So the label
of a terminated
+ resource, e.g. `Failed`, stays as it was. To move a running resource to
another queue, set
+ `spec.suspend` to `true`, change the label once the resource is `Suspended`,
and then resume it,
+ while a running `SparkApplication` also moves with its next attempt after a
restart. Before a
+ resource starts, the operator follows the label as described
above. The policy reads only the status, so a resource whose first status is
not persisted yet,
e.g. after the update following the request of its driver or master failed,
looks like it has not
started. Its label may still change then, while the operator keeps the
admitted `Workload` of such
@@ -720,11 +747,11 @@ spec:
* Once the integration is disabled or the access of the operator to
`Workload`s is revoked, e.g.
by disabling `operatorRbac.kueue.enabled`, the operator no longer deletes
them, and a `Workload`
left behind keeps its quota until its owner is deleted. While the
integration stays enabled
- without the access, a suspended `SparkCluster` does not resume, and a
stopping
- `SparkApplication` neither terminates nor restarts, until the access is
restored or the
+ without the access, a suspended `SparkApplication` or `SparkCluster` does
not resume, and a
+ stopping `SparkApplication` neither terminates nor restarts, until the
access is restored or the
integration is disabled, since the operator keeps deleting their `Workload`s
by name, which is
denied even once a `Workload` is gone. Before disabling either, let the
queued
- `SparkApplication`s finish or suspend those which have not started yet, and
suspend the queued
+ `SparkApplication`s finish or suspend them, and suspend the queued
`SparkCluster`s, so that the operator releases their `Workload`s itself.
Then list the remaining
ones with `kubectl get workloads -A -l spark.operator/spark-app-name` and
`kubectl get workloads -A -l spark.operator/spark-cluster-name`, and delete
only those whose
@@ -790,11 +817,12 @@ spec:
is a `PodsReadyTimeout`, e.g. of an application which runs with fewer
executors than
`spark.executor.instances`, or a deactivation. The operator checks the
admission only before it
creates the resources, so after a later eviction the driver and executors
keep running, and
- until the attempt stops, the `KueueEvictionIgnored`
[event](configuration.md#kubernetes-events)
- is published instead by default. An evicted `Workload` keeps its Kueue quota
until the driver
- and executors are released, so a workload which preempts the application
waits until then,
- while a deactivated one no longer counts against it. Since `spec.suspend`
does not stop a
- running attempt, delete the application to release its quota earlier. With
`blockAdmission`, an
+ until the attempt stops or is suspended, the `KueueEvictionIgnored`
+ [event](configuration.md#kubernetes-events) is published instead by default.
An evicted
+ `Workload` keeps its Kueue quota until the driver and executors are
released, so a workload
+ which preempts the application waits until then, while a deactivated one no
longer counts
+ against it. Set `spec.suspend` to `true`, as described in
[Suspend](#suspend), or delete the
+ application to release its quota earlier. With `blockAdmission`, an
application whose pods are not all ready holds back every other workload
until they are ready
or its quota is released as described above.
* To limit the execution time of a `SparkApplication`, the Spark native
`spark.driver.timeout` is
diff --git
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
index de5eaa87..7286a29e 100644
---
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
+++
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java
@@ -230,6 +230,22 @@ public final class Constants {
public static final String EXECUTOR_LAUNCH_TIMEOUT_MESSAGE =
"The Spark application failed to get enough executors in the given time
threshold.";
+ /**
+ * Message indicating that the application is suspended by spec.suspend
after its driver was
+ * requested.
+ */
+ public static final String APP_SUSPENDED_MESSAGE =
+ "Application is suspended by spec.suspend, so its driver and executors
are being released. "
+ + "Setting spec.suspend to false resumes it with a new attempt once
they are released.";
+
+ /**
+ * Message indicating that the suspended application is resumed by clearing
spec.suspend, with a
+ * new attempt.
+ */
+ public static final String APP_RESUMED_MESSAGE =
+ "Application is resumed as spec.suspend is set to false, so a new
attempt starts from "
+ + "scratch.";
+
// Spark Cluster Messages
/** Message indicating a failure to request the Spark cluster from the
scheduler backend. */
public static final String CLUSTER_SCHEDULE_FAILURE_MESSAGE =
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
index 39e2ccf5..2cb9ec3f 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
@@ -264,11 +264,11 @@ public final class SparkOperatorConf {
/**
* Requeue interval (in seconds) at which a resource held by {@code
spec.suspend} is reconciled,
- * so that its {@code SuspendHeld} event is republished. A 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
- * {@code spec.suspend}, which arrives as a watch event and reconciles right
away, so nothing
- * waits for this interval. It is deliberately much coarser than {@link
+ * so that its {@code 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 {@code spec.suspend}, which arrives as a watch event and
reconciles right away, so
+ * nothing waits for this interval. It is deliberately much coarser than
{@link
* #RECONCILER_INTERVAL_SECONDS}, 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
{@code --event-ttl} of
* the API server (one hour by default), or the event expires between
repeats and the hold stops
@@ -282,8 +282,9 @@ public final class SparkOperatorConf {
.enableDynamicOverride(true)
.description(
"Requeue interval (in seconds) at which a resource held by
spec.suspend is "
- + "reconciled, so that its SuspendHeld event is republished.
A SparkCluster "
- + "suspended while running publishes no such event, since
its Suspended status "
+ + "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 "
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
index 24ac7629..9e8cbf53 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java
@@ -59,6 +59,7 @@ import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppKueueEvictionS
import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppReconcileStep;
import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppResourceObserveStep;
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppRunningStep;
+import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppSuspendStep;
import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppUnknownStateStep;
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppValidateStep;
import org.apache.spark.k8s.operator.utils.EventUtils;
@@ -212,6 +213,7 @@ public class SparkAppReconciler implements
Reconciler<SparkApplication>, Cleaner
List.of(new AppDriverStartObserver(), new
AppDriverReadyObserver())));
steps.add(new AppResourceObserveStep(List.of(new
AppDriverRunningObserver())));
steps.add(new AppResourceObserveStep(List.of(new
AppDriverTimeoutObserver())));
+ steps.add(new AppSuspendStep());
}
case DriverReady,
InitializedBelowThresholdExecutors,
@@ -222,7 +224,9 @@ public class SparkAppReconciler implements
Reconciler<SparkApplication>, Cleaner
steps.add(new AppRunningStep());
steps.add(new AppResourceObserveStep(List.of(new
AppDriverRunningObserver())));
steps.add(new AppResourceObserveStep(List.of(new
AppDriverTimeoutObserver())));
+ steps.add(new AppSuspendStep());
}
+ case Suspended -> steps.add(new AppSuspendStep());
default -> steps.add(new AppUnknownStateStep());
}
return steps;
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
index b1bf58c8..e03851ec 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
@@ -91,7 +91,10 @@ public final class AppInitStep extends AppReconcileStep {
return SuspendUtils.holdForSuspend(context, "driver");
}
}
- if (app.getStatus().getPreviousAttemptSummary() != null) {
+ // Only a restarted attempt waits for the restart backoff, while the
attempt which a suspended
+ // application starts on resume is Submitted.
+ if (ApplicationStateSummary.ScheduledToRestart ==
currentState.getCurrentStateSummary()
+ && app.getStatus().getPreviousAttemptSummary() != null) {
Instant lastTransitionTime =
Instant.parse(currentState.getLastTransitionTime());
ApplicationAttemptSummary attemptSummary =
app.getStatus().getPreviousAttemptSummary();
SortedMap<Long, ApplicationState> attemptHistory =
attemptSummary.getStateTransitionHistory();
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStep.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStep.java
index cbeb760e..26439d41 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStep.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStep.java
@@ -78,7 +78,7 @@ public final class AppKueueEvictionStep extends
AppReconcileStep {
+ cause
+ ", which the operator does not act on once the driver is
requested, so the "
+ "driver and executors keep running and holding its quota
until they are "
- + "released. Delete the application to release them.");
+ + "released. Set spec.suspend to true or delete the
application to release them.");
return proceed();
}
}
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
index cb1d9db2..caea2e58 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppReconcileStep.java
@@ -45,7 +45,7 @@ import
org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
@Log4j2
public abstract sealed class AppReconcileStep
permits AppCleanUpStep, AppInitStep, AppKueueEvictionStep,
AppResourceObserveStep,
- AppRunningStep, AppUnknownStateStep, AppValidateStep {
+ AppRunningStep, AppSuspendStep, AppUnknownStateStep, AppValidateStep {
/**
* Reconciles a specific step for a Spark application.
*
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStep.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStep.java
new file mode 100644
index 00000000..9737b743
--- /dev/null
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStep.java
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.reconciler.reconcilesteps;
+
+import static org.apache.spark.k8s.operator.Constants.APP_RESUMED_MESSAGE;
+import static org.apache.spark.k8s.operator.Constants.APP_SUSPENDED_MESSAGE;
+import static
org.apache.spark.k8s.operator.Constants.LABEL_SPARK_APPLICATION_NAME;
+import static
org.apache.spark.k8s.operator.Constants.LABEL_SPARK_ROLE_DRIVER_VALUE;
+import static
org.apache.spark.k8s.operator.Constants.LABEL_SPARK_ROLE_EXECUTOR_VALUE;
+import static org.apache.spark.k8s.operator.Constants.LABEL_SPARK_ROLE_NAME;
+import static
org.apache.spark.k8s.operator.config.SparkOperatorConf.SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS;
+import static
org.apache.spark.k8s.operator.config.SparkOperatorConf.TRIM_ATTEMPT_STATE_TRANSITION_HISTORY;
+import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndDefaultRequeue;
+import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndImmediateRequeue;
+import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndRequeueAfter;
+import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.proceed;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import io.fabric8.kubernetes.api.model.DeletionPropagation;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.spark.k8s.operator.SparkApplication;
+import org.apache.spark.k8s.operator.context.SparkAppContext;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
+import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import
org.apache.spark.k8s.operator.reconciler.observers.AppDriverRunningObserver;
+import org.apache.spark.k8s.operator.spec.ApplicationTimeoutConfig;
+import org.apache.spark.k8s.operator.status.ApplicationState;
+import org.apache.spark.k8s.operator.status.ApplicationStateSummary;
+import org.apache.spark.k8s.operator.utils.EventUtils;
+import org.apache.spark.k8s.operator.utils.ReconcilerUtils;
+import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
+
+/**
+ * Suspends an application whose driver is requested by spec.suspend, and
resumes it. An
+ * application suspended before its driver is requested is held by {@link
AppInitStep} instead.
+ *
+ * <p>An application enters Suspended before anything is released, and its
driver is released only
+ * in Suspended. Releasing it first would leave an application without its
driver in a running state
+ * when spec.suspend is cleared in the meantime, and the driver observers
would fail it. A resumed
+ * application leaves Suspended only after everything is released, and then
starts a new attempt
+ * from Submitted, where {@link AppInitStep} requests its driver and a new
Kueue admission. Resuming
+ * earlier would let the new attempt run on the admitted Workload of the
suspended one, whose name
+ * is fixed per application, or meet the driver resources of the suspended
one, whose names stay
+ * the same if spark.app.id is set. Like a restart, the new attempt gets the
next attempt id, while
+ * it does not count against the restart limits, see {@link
+ * org.apache.spark.k8s.operator.status.ApplicationStatus#resume}.
+ *
+ * <p>It runs after the driver is observed, so that an attempt whose driver
completed or failed
+ * meanwhile ends as usual, rather than being suspended and run again from
scratch on resume. The
+ * observers read the informer cache, which may not show such a driver yet, so
the driver is
+ * observed once more on the API server before the application is suspended.
+ */
+@Slf4j
+public final class AppSuspendStep extends AppReconcileStep {
+ /** The states in which the driver is requested, and the attempt has not
stopped. */
+ private static final Set<ApplicationStateSummary> DRIVER_REQUESTED_STATES =
+ EnumSet.range(
+ ApplicationStateSummary.DriverRequested,
+ ApplicationStateSummary.RunningWithBelowThresholdExecutors);
+
+ /**
+ * Reconciles an application whose driver is requested, or which is
suspended, against
+ * spec.suspend.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status
updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
+ @Override
+ public ReconcileProgress reconcile(
+ SparkAppContext context, SparkAppStatusRecorder statusRecorder) {
+ SparkApplication app = context.getResource();
+ boolean suspend = app.getSpec().isSuspend();
+ ApplicationStateSummary summary =
app.getStatus().getCurrentState().getCurrentStateSummary();
+ if (DRIVER_REQUESTED_STATES.contains(summary) && suspend) {
+ Optional<ApplicationState> termination;
+ try {
+ termination = observeDriverTermination(context);
+ } catch (KubernetesClientException e) {
+ log.warn("Failed to look up the driver of the application to suspend,
will retry.", e);
+ return completeAndDefaultRequeue();
+ }
+ if (termination.isPresent()) {
+ return appendStateAndImmediateRequeue(context, statusRecorder,
termination.get());
+ }
+ // A rejected update, e.g. of a state unknown to an older CRD, would be
rejected again right
+ // away, so it is retried with the default interval, not immediately.
+ return statusRecorder.appendNewStateAndPersist(
+ context,
+ new ApplicationState(ApplicationStateSummary.Suspended,
APP_SUSPENDED_MESSAGE))
+ ? completeAndImmediateRequeue()
+ : completeAndDefaultRequeue();
+ }
+ if (summary != ApplicationStateSummary.Suspended) {
+ return proceed();
+ }
+ Optional<ReconcileProgress> waiting = releaseResources(context);
+ if (waiting.isPresent()) {
+ return waiting.get();
+ }
+ if (suspend) {
+ return completeAndRequeueAfter(
+
Duration.ofSeconds(SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS.getValue()));
+ }
+ return statusRecorder.persistStatus(
+ context,
+ app.getStatus()
+ .resume(APP_RESUMED_MESSAGE,
TRIM_ATTEMPT_STATE_TRANSITION_HISTORY.getValue()))
+ ? completeAndImmediateRequeue()
+ : completeAndDefaultRequeue();
+ }
+
+ /**
+ * Observes whether the driver of the attempt completed or failed, as the
API server knows it. A
+ * driver which is being deleted, e.g. one of an earlier attempt, is not
looked at, see {@link
+ * SparkAppContext#isLiveDriverPod}.
+ *
+ * @param context The SparkAppContext for the application.
+ * @return The state which ends the attempt if its driver completed or
failed, or empty.
+ * @throws KubernetesClientException if the driver pods cannot be listed.
+ */
+ private static Optional<ApplicationState>
observeDriverTermination(SparkAppContext context) {
+ SparkApplication app = context.getResource();
+ AppDriverRunningObserver observer = new AppDriverRunningObserver();
+ return listPods(context, LABEL_SPARK_ROLE_DRIVER_VALUE).stream()
+ .filter(pod -> SparkAppContext.isLiveDriverPod(app, pod))
+ .map(pod -> observer.observe(pod, app.getSpec(), app.getStatus()))
+ .flatMap(Optional::stream)
+ .findFirst();
+ }
+
+ /**
+ * Releases the driver of a suspended application, and then its Kueue
Workload. Like at the end
+ * of an attempt, only the driver pod is deleted, whatever the
resourceRetainPolicy is, and its
+ * executors and the resources which it owns are deleted with it. The pods
are listed on the API
+ * server, so that every driver pod is deleted, including one which the
informer cache lost, or
+ * one besides a driver of an earlier attempt which is still terminating.
Unlike at the end of an
+ * attempt, the deletion is not waited for here, since it is observed by the
pod informer, which
+ * reconciles again.
+ *
+ * <p>The Workload is released only after the driver and executor pods are
gone, since Kueue would
+ * admit another workload into the quota which the terminating pods still
occupy, even if the
+ * queue label was removed after the admission. Every application waits for
its pods, with or
+ * without a Workload, since the resumed attempt starts without a restart
backoff, and its driver
+ * observers would take a driver of the suspended attempt which is still
terminating for theirs.
+ * A pod which is still terminating `forceTerminationGracePeriodMillis`
after its grace period
+ * ended, e.g. on a lost node, no longer holds the application, and a driver
pod is force deleted
+ * then, so that it does not keep its pod object. Like for a suspended
SparkCluster, it is
+ * measured per pod rather than since Suspended, so that a release which
failed or was delayed
+ * for that long still waits for the pods it deletes.
+ *
+ * <p>Everything here is idempotent, so pods that are not gone yet, or a
release that failed, are
+ * simply looked at again. A failure which is not expected to clear on its
own is reported, since
+ * Suspended is a steady state which would not tell it apart from waiting
for the pods to go.
+ *
+ * @param context The SparkAppContext for the application.
+ * @return Empty once everything is released, or the ReconcileProgress to
retry with while pods
+ * remain or a release failed.
+ */
+ private Optional<ReconcileProgress> releaseResources(SparkAppContext
context) {
+ SparkApplication app = context.getResource();
+ ApplicationTimeoutConfig timeoutConfig =
+
app.getSpec().getApplicationTolerations().getApplicationTimeoutConfig();
+ KubernetesClient client = context.getClient();
+ try {
+ List<Pod> pods =
+ listPods(context, LABEL_SPARK_ROLE_DRIVER_VALUE,
LABEL_SPARK_ROLE_EXECUTOR_VALUE);
+ Instant now = Instant.now();
+ boolean waiting = false;
+ Instant nextDeadline = Instant.MAX;
+ for (Pod pod : pods) {
+ Instant deadline = getPodReleaseDeadline(pod, timeoutConfig);
+ boolean holding = now.isBefore(deadline);
+ boolean driver =
+ LABEL_SPARK_ROLE_DRIVER_VALUE.equals(
+ pod.getMetadata().getLabels().get(LABEL_SPARK_ROLE_NAME));
+ if (driver && !holding) {
+ client.resource(pod).withGracePeriod(0L).delete();
+ } else if (driver && pod.getMetadata().getDeletionTimestamp() == null)
{
+ // A driver which is terminating already is not deleted again until
it is force deleted
+
client.resource(pod).withPropagationPolicy(DeletionPropagation.FOREGROUND).delete();
+ }
+ if (holding && deadline.isBefore(nextDeadline)) {
+ nextDeadline = deadline;
+ }
+ waiting |= holding;
+ }
+ if (waiting) {
+ // The deletion of each pod is observed by the pod informer, while the
end of the wait for
+ // a pod which may send no more events is observed by the requeue
+ log.debug("Waiting for the driver and executor pods of the suspended
app to be deleted.");
+ ReconcileProgress defaultRequeue = completeAndDefaultRequeue();
+ Duration remaining = Duration.between(now, nextDeadline);
+ return Optional.of(
+ remaining.compareTo(defaultRequeue.getRequeueAfterDuration()) < 0
+ ? completeAndRequeueAfter(remaining)
+ : defaultRequeue);
+ }
+ // A Workload admitted before the queue label was removed is released as
well.
+ KueueWorkloadUtils.deleteWorkloadOf(client, app);
+ } catch (KubernetesClientException e) {
+ log.warn("Failed to release the resources of the suspended application,
will retry.", e);
+ if (!ReconcilerUtils.isTransientError(e)) {
+ EventUtils.warn(
+ context.getEventRecorder(),
+ EventUtils.REASON_SUSPEND_RELEASE_FAILED,
+ "Failed to release the driver and executors or the Kueue Workload
of the suspended "
+ + "application, will retry. "
+ + EventUtils.describe(e));
+ }
+ return Optional.of(completeAndDefaultRequeue());
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Lists the pods of the application with the given Spark roles on the API
server, including the
+ * ones which are being deleted. Other pods which carry the application
label do not count.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param roles The values of the Spark role label of the pods to list.
+ * @return The pods of the application with the given roles.
+ * @throws KubernetesClientException if the pods cannot be listed.
+ */
+ private static List<Pod> listPods(SparkAppContext context, String... roles) {
+ SparkApplication app = context.getResource();
+ return ReconcilerUtils.podsOf(
+ context.getClient(),
+ app.getMetadata().getNamespace(),
+ LABEL_SPARK_APPLICATION_NAME,
+ app.getMetadata().getName(),
+ roles)
+ .list()
+ .getItems();
+ }
+
+ /**
+ * Returns when a driver or executor pod stops holding a suspended
application, and a driver pod
+ * is force deleted.
+ *
+ * @param pod The driver or executor pod.
+ * @param timeoutConfig The ApplicationTimeoutConfig of the application.
+ * @return `forceTerminationGracePeriodMillis` after its deletionTimestamp,
when its grace period
+ * ends, or Instant.MAX while it is not being deleted yet.
+ */
+ private static Instant getPodReleaseDeadline(Pod pod,
ApplicationTimeoutConfig timeoutConfig) {
+ String deletionTimestamp = pod.getMetadata().getDeletionTimestamp();
+ return deletionTimestamp == null
+ ? Instant.MAX
+ : Instant.parse(deletionTimestamp)
+ .plusMillis(timeoutConfig.getForceTerminationGracePeriodMillis());
+ }
+}
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java
index 86f684fe..c303d681 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/SuspendUtils.java
@@ -50,8 +50,9 @@ final class SuspendUtils {
* of the one Event instead of creating another and refreshes it, so that
the hold stays visible
* past the event retention of the API server. A suspended first attempt has
no persisted status
* to fall back on, since its initial Submitted state is never written. An
application held
- * later, in ScheduledToRestart, does have the status of the previous
attempt, but that status
- * does not say that the next attempt is withheld by spec.suspend, so it
gets the same event.
+ * later, in ScheduledToRestart, or in Submitted after it was resumed from
Suspended, does have a
+ * persisted status, but that status does not say that the next attempt is
withheld by
+ * spec.suspend, so it gets the same event.
*
* <p>Unlike the Kueue hold, which ends when quota arrives, this one ends
only when a user clears
* spec.suspend, so the republishing is paced by {@link
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java
index a33384f3..376fa3a3 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java
@@ -68,8 +68,8 @@ public final class EventUtils {
public static final String REASON_SUSPEND_HELD = "SuspendHeld";
/**
- * Reason for an event describing a failure to release the resources of a
SparkCluster which is
- * suspended while running.
+ * Reason for an event describing a failure to release the resources of a
SparkApplication or a
+ * SparkCluster which is suspended while running.
*/
public static final String REASON_SUSPEND_RELEASE_FAILED =
"SuspendReleaseFailed";
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
index 572645dc..fc6ad9b5 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java
@@ -71,6 +71,8 @@ import
io.javaoperatorsdk.operator.processing.event.ResourceID;
import
io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
@@ -85,9 +87,11 @@ import
org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
import org.apache.spark.k8s.operator.kueue.v1beta2.Workload;
import org.apache.spark.k8s.operator.kueue.v1beta2.WorkloadStatus;
import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager;
+import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppCleanUpStep;
import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppKueueEvictionStep;
import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppReconcileStep;
-import
org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppResourceObserveStep;
+import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppSuspendStep;
+import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppValidateStep;
import org.apache.spark.k8s.operator.spec.ApplicationSpec;
import org.apache.spark.k8s.operator.spec.ApplicationTolerations;
import org.apache.spark.k8s.operator.spec.RestartConfig;
@@ -102,6 +106,11 @@ import
org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
import org.apache.spark.k8s.operator.utils.Utils;
class SparkAppReconcilerTest {
+ private static final Set<ApplicationStateSummary> DRIVER_REQUESTED_STATES =
+ EnumSet.range(
+ ApplicationStateSummary.DriverRequested,
+ ApplicationStateSummary.RunningWithBelowThresholdExecutors);
+
private final SparkAppStatusRecorder mockRecorder =
mock(SparkAppStatusRecorder.class);
private final SentinelManager<SparkApplication> mockSentinelManager =
mock(SentinelManager.class);
private final KubernetesClient mockClient = mock(KubernetesClient.class);
@@ -529,10 +538,13 @@ class SparkAppReconcilerTest {
}
}
- @Test
- void
observeStepsRecordDriverCompletedBeforeFirstObservationAsSucceededOnly() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void stepsRecordDriverCompletedBeforeFirstObservationAsSucceededOnly(boolean
suspend) {
app.setMetadata(new
ObjectMetaBuilder().withName("app").withNamespace("default").build());
app.setSpec(new ApplicationSpec());
+ // Even if suspended, the attempt ends as usual rather than being run
again on resume
+ app.getSpec().setSuspend(suspend);
Pod driver =
new PodBuilder()
.withNewMetadata()
@@ -565,14 +577,16 @@ class SparkAppReconcilerTest {
});
for (ApplicationStateSummary summary :
- List.of(ApplicationStateSummary.DriverRequested,
ApplicationStateSummary.DriverStarted)) {
+ List.of(
+ ApplicationStateSummary.DriverRequested,
+ ApplicationStateSummary.DriverStarted,
+ ApplicationStateSummary.RunningHealthy)) {
app.setStatus(new ApplicationStatus().appendNewState(new
ApplicationState(summary, "")));
int previousSize = app.getStatus().getStateTransitionHistory().size();
SparkAppContext context = new SparkAppContext(app, mockContext,
mockWorker);
- // Run the observe steps selected by the reconciler until one completes
the reconciliation
+ // Run the steps selected by the reconciler until one completes the
reconciliation
for (AppReconcileStep step : reconciler.getReconcileSteps(app)) {
- if (step instanceof AppResourceObserveStep
- && step.reconcile(context, recorder).isCompleted()) {
+ if (step.reconcile(context, recorder).isCompleted()) {
break;
}
}
@@ -586,16 +600,36 @@ class SparkAppReconcilerTest {
}
}
+ @Test
+ void suspendStepRunsLastOnceTheDriverIsRequestedAndAloneWhileSuspended() {
+ for (ApplicationStateSummary summary : ApplicationStateSummary.values()) {
+ app.setStatus(new ApplicationStatus().appendNewState(new
ApplicationState(summary, "")));
+ List<AppReconcileStep> steps = reconciler.getReconcileSteps(app);
+ if (summary == ApplicationStateSummary.Suspended) {
+ // Not handled as an unknown state, which would fail the application
+ assertEquals(
+ List.of(AppValidateStep.class, AppCleanUpStep.class,
AppSuspendStep.class),
+ steps.stream().map(Object::getClass).toList());
+ } else {
+ boolean expected = DRIVER_REQUESTED_STATES.contains(summary);
+ // After the observers, so that a driver which completed meanwhile is
recorded rather than
+ // suspended and run again when the application is resumed
+ assertEquals(
+ expected, steps.get(steps.size() - 1) instanceof AppSuspendStep,
summary.name());
+ assertEquals(
+ expected ? 1 : 0,
+ steps.stream().filter(AppSuspendStep.class::isInstance).count(),
+ summary.name());
+ }
+ }
+ }
+
@Test
void kueueEvictionIsReportedFromTheDriverRequestUntilTheAttemptStops() {
- Set<ApplicationStateSummary> driverRequested =
- EnumSet.range(
- ApplicationStateSummary.DriverRequested,
- ApplicationStateSummary.RunningWithBelowThresholdExecutors);
for (ApplicationStateSummary summary : ApplicationStateSummary.values()) {
app.setStatus(new ApplicationStatus().appendNewState(new
ApplicationState(summary, "")));
List<AppReconcileStep> steps = reconciler.getReconcileSteps(app);
- boolean expected = driverRequested.contains(summary);
+ boolean expected = DRIVER_REQUESTED_STATES.contains(summary);
// Right after the validation and the clean up, before a state
transition ends the
// reconciliation. Before the driver is requested, AppInitStep handles
the eviction itself.
assertEquals(expected, steps.get(2) instanceof AppKueueEvictionStep,
summary.name());
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java
index 5bcc3db6..ca185ff3 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java
@@ -409,6 +409,42 @@ class AppInitStepTest {
application.getStatus().getCurrentState().getCurrentStateSummary());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void resumedAttemptDoesNotWaitForRestartBackoff(boolean
trimStateTransitionHistory) {
+ // An attempt which starts as the application is resumed from Suspended is
not a restart
+ AppInitStep appInitStep = new AppInitStep();
+ SparkAppContext mockContext = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+ SparkApplication application = new SparkApplication();
+ application.setMetadata(applicationMetadata);
+ application.getSpec().setApplicationTolerations(
+ ApplicationTolerations.builder()
+
.restartConfig(RestartConfig.builder().restartBackoffMillis(60000L).build())
+ .build());
+ application.setStatus(
+ new ApplicationStatus()
+ .appendNewState(new
ApplicationState(ApplicationStateSummary.DriverRequested, ""))
+ .appendNewState(new
ApplicationState(ApplicationStateSummary.Suspended, ""))
+ .resume("resumed", trimStateTransitionHistory));
+ when(mockContext.getResource()).thenReturn(application);
+ when(mockContext.getDriverPreResourcesSpec()).thenReturn(List.of());
+ when(mockContext.getDriverPodSpec()).thenReturn(driverPodSpec);
+ when(mockContext.getDriverResourcesSpec()).thenReturn(List.of());
+ when(mockContext.getClient()).thenReturn(kubernetesClient);
+ when(recorder.persistStatus(any(), any())).thenAnswer(invocation -> {
+ application.setStatus(invocation.getArgument(1));
+ return true;
+ });
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ appInitStep.reconcile(mockContext, recorder));
+ Assertions.assertEquals(
+ ApplicationStateSummary.DriverRequested,
+ application.getStatus().getCurrentState().getCurrentStateSummary());
+ }
+
@Test
void suspendedAppDoesNotRequestDriver() {
AppInitStep appInitStep = new AppInitStep();
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStepTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStepTest.java
index fac5276b..810b656e 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStepTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStepTest.java
@@ -87,6 +87,7 @@ class AppKueueEvictionStepTest {
// Kueue keeps counting the quota of an evicted Workload until the driver
and executors are
// released
Assertions.assertTrue(message.contains("holding its quota"), message);
+ Assertions.assertTrue(message.contains("Set spec.suspend to true"),
message);
verifyNoInteractions(recorder);
}
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStepTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStepTest.java
new file mode 100644
index 00000000..5d8cde58
--- /dev/null
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppSuspendStepTest.java
@@ -0,0 +1,729 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.reconciler.reconcilesteps;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import io.fabric8.mockwebserver.http.RecordedRequest;
+import io.javaoperatorsdk.operator.api.event.EventRecord;
+import io.javaoperatorsdk.operator.api.event.EventType;
+import io.javaoperatorsdk.operator.api.event.ResourceEventRecorder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+
+import org.apache.spark.k8s.operator.Constants;
+import org.apache.spark.k8s.operator.SparkApplication;
+import org.apache.spark.k8s.operator.config.SparkOperatorConf;
+import org.apache.spark.k8s.operator.context.SparkAppContext;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadFactory;
+import org.apache.spark.k8s.operator.kueue.v1beta2.Workload;
+import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import org.apache.spark.k8s.operator.spec.ResourceRetainPolicy;
+import org.apache.spark.k8s.operator.status.ApplicationState;
+import org.apache.spark.k8s.operator.status.ApplicationStateSummary;
+import org.apache.spark.k8s.operator.status.ApplicationStatus;
+import org.apache.spark.k8s.operator.utils.EventUtils;
+import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
+import org.apache.spark.k8s.operator.utils.TestUtils;
+import org.apache.spark.k8s.operator.utils.Utils;
+
+@EnableKubernetesMockClient(crud = true)
+class AppSuspendStepTest {
+ private static final String DRIVER = "app1-0-driver";
+ private static final String EXECUTOR = "app1-0-exec-1";
+ // The driver of the attempt after the one of DRIVER
+ private static final String NEXT_DRIVER = "app1-1-driver";
+ // Keeps a deleted pod in the mock server, like the grace period of a
terminating pod
+ private static final String FINALIZER = "example.com/finalizer";
+
+ private KubernetesMockServer server;
+ private KubernetesClient kubernetesClient;
+
+ // The default of
spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds
+ private static final ReconcileProgress SUSPEND_HOLD_PROGRESS =
+ ReconcileProgress.completeAndRequeueAfter(Duration.ofMinutes(30));
+
+ // The deletion of each pod is observed by the pod informer, which
reconciles again
+ private static final ReconcileProgress WAITING_FOR_PODS_PROGRESS =
+ ReconcileProgress.completeAndDefaultRequeue();
+
+ private final SparkAppContext mockContext = mock(SparkAppContext.class);
+ private final SparkAppStatusRecorder recorder =
mock(SparkAppStatusRecorder.class);
+ private final ResourceEventRecorder eventRecorder =
mock(ResourceEventRecorder.class);
+
+ @BeforeEach
+ void enableKueue() {
+ TestUtils.setConfigKey(SparkOperatorConf.KUEUE_ENABLED, true);
+ }
+
+ @AfterEach
+ void disableKueue() {
+ TestUtils.setConfigKey(SparkOperatorConf.KUEUE_ENABLED, false);
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = ApplicationStateSummary.class,
+ from = "DriverRequested",
+ to = "RunningWithBelowThresholdExecutors")
+ void runningAppWithoutSuspendProceeds(ApplicationStateSummary summary) {
+ SparkApplication app = buildApp(summary, false);
+ stubContext(app);
+ createDriver(app);
+
+ Assertions.assertEquals(
+ ReconcileProgress.proceed(), new
AppSuspendStep().reconcile(mockContext, recorder));
+
+ verifyNoInteractions(recorder);
+ Assertions.assertNotNull(getDriver());
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = ApplicationStateSummary.class,
+ from = "DriverRequested",
+ to = "RunningWithBelowThresholdExecutors")
+ void suspendingRunningAppEntersSuspendedBeforeReleasingResources(
+ ApplicationStateSummary summary) {
+ SparkApplication app = buildKueueApp(summary, true);
+ stubContext(app);
+ createDriver(app);
+ createWorkload(app);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ ApplicationState state = captureAppendedState();
+ Assertions.assertEquals(ApplicationStateSummary.Suspended,
state.getCurrentStateSummary());
+ Assertions.assertEquals(Constants.APP_SUSPENDED_MESSAGE,
state.getMessage());
+ // Nothing is released until Suspended is persisted, so that an
application which is resumed
+ // in the meantime is never left running without its driver, which would
fail it.
+ Assertions.assertNotNull(getDriver());
+ Assertions.assertNotNull(getWorkload());
+ }
+
+ @Test
+ void suspendedStateWhichIsNotPersistedIsRetriedLater() {
+ SparkApplication app = buildApp(ApplicationStateSummary.RunningHealthy,
true);
+ stubContext(app);
+ createDriver(app);
+ when(recorder.appendNewStateAndPersist(any(), any())).thenReturn(false);
+
+ // A status which is rejected, e.g. by a CRD without the Suspended state,
is not retried at once
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ Assertions.assertNotNull(getDriver());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"Succeeded", "Failed"})
+ void attemptWhoseDriverEndedMeanwhileEndsAsUsual(String phase) {
+ SparkApplication app =
buildKueueApp(ApplicationStateSummary.RunningHealthy, true);
+ stubContext(app);
+ // The API server knows that the driver ended, while the informer cache
which the driver
+ // observers read may not show it yet
+ kubernetesClient
+ .resource(new
PodBuilder(driver(app)).withNewStatus().withPhase(phase).endStatus().build())
+ .create();
+ createWorkload(app);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ // The attempt ends rather than being suspended and run again from scratch
on resume
+ Assertions.assertEquals(
+ ApplicationStateSummary.valueOf(phase),
captureAppendedState().getCurrentStateSummary());
+ Assertions.assertNotNull(getDriver());
+ Assertions.assertNotNull(getWorkload());
+ }
+
+ @Test
+ void terminatingDriverOfEarlierAttemptDoesNotEndTheAttempt() {
+ SparkApplication app = buildApp(ApplicationStateSummary.RunningHealthy,
true);
+ stubContext(app);
+ // The failed driver of an earlier attempt, which is still terminating
+ kubernetesClient
+ .resource(
+ withFinalizers(
+ new
PodBuilder(driver(app)).withNewStatus().withPhase("Failed").endStatus().build(),
+ FINALIZER))
+ .create();
+ kubernetesClient.resource(driver(app)).delete();
+ kubernetesClient.resource(pod(NEXT_DRIVER,
Utils.driverLabels(app))).create();
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ Assertions.assertEquals(
+ ApplicationStateSummary.Suspended,
captureAppendedState().getCurrentStateSummary());
+ }
+
+ @Test
+ void suspensionWaitsUntilTheDriverCanBeLookedUp() {
+ SparkApplication app = buildApp(ApplicationStateSummary.RunningHealthy,
true);
+ KubernetesClient client = spy(kubernetesClient);
+ doThrow(new KubernetesClientException("Service Unavailable", 503,
null)).when(client).pods();
+ stubContext(app, client);
+
+ // Whether the driver ended meanwhile is unknown, so the application is
not suspended yet
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ verifyNoInteractions(recorder, eventRecorder);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void suspendedAppKeepsKueueWorkloadWhilePodsRemain(boolean suspend) throws
Exception {
+ // Resuming waits as well, so that the new attempt does not run on the
admission of the
+ // suspended one
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
suspend);
+ stubContext(app);
+ createDriver(app, FINALIZER);
+ createWorkload(app);
+
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+
+ // The driver is deleted with its executors and the resources it owns,
without forcing it
+ // within the grace period
+ List<String> deletions = podDeleteOptions();
+ Assertions.assertEquals(1, deletions.size(), deletions.toString());
+
Assertions.assertTrue(deletions.get(0).contains("\"propagationPolicy\":\"Foreground\""));
+
Assertions.assertFalse(deletions.get(0).contains("\"gracePeriodSeconds\":0"));
+ Assertions.assertNotNull(getDriver().getMetadata().getDeletionTimestamp());
+ // The driver is still terminating, so it keeps holding the quota of the
Workload
+ Assertions.assertNotNull(getWorkload());
+
+ // The terminating driver is not deleted again while it is waited for
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+ Assertions.assertEquals(List.of(), podDeleteOptions());
+ verifyNoInteractions(recorder, eventRecorder);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void podsOfAppWithoutKueueAreWaitedForAsWell(boolean suspend) {
+ // A resumed application starts its new attempt without a restart backoff,
and the observers
+ // of that attempt would take a driver of the suspended one for theirs
+ TestUtils.setConfigKey(SparkOperatorConf.KUEUE_ENABLED, false);
+ SparkApplication app = buildApp(ApplicationStateSummary.Suspended,
suspend);
+ stubContext(app);
+ createDriver(app);
+ createExecutor(app);
+
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+
+ Assertions.assertNull(getDriver());
+ verifyNoInteractions(recorder);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void podsAreWaitedForWhenTheReleaseStartsLate(boolean suspend) throws
Exception {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
suspend);
+ // Suspended for longer than forceTerminationGracePeriodMillis, 5 minutes
by default, e.g. as
+ // the operator was down meanwhile or the deletion kept failing
+ app.getStatus()
+ .getCurrentState()
+
.setLastTransitionTime(Instant.now().minus(Duration.ofMinutes(6)).toString());
+ stubContext(app);
+ createDriver(app, FINALIZER);
+ createExecutor(app);
+ createWorkload(app);
+
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+
+ // The wait is measured per pod, so the driver still gets its grace
period, and the quota which
+ // its executors occupy is not given back yet
+ List<String> deletions = podDeleteOptions();
+ Assertions.assertEquals(1, deletions.size(), deletions.toString());
+
Assertions.assertTrue(deletions.get(0).contains("\"propagationPolicy\":\"Foreground\""));
+
Assertions.assertFalse(deletions.get(0).contains("\"gracePeriodSeconds\":0"));
+ Assertions.assertNotNull(getWorkload());
+ verifyNoInteractions(recorder, eventRecorder);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void liveDriverIsReleasedBesideTerminatingDriverOfEarlierAttempt(boolean
suspend)
+ throws Exception {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
suspend);
+ stubContext(app);
+ // The driver of an earlier attempt, which is still terminating, e.g. on a
lost node
+ createDriver(app, FINALIZER);
+ kubernetesClient.resource(driver(app)).delete();
+ kubernetesClient.resource(pod(NEXT_DRIVER,
Utils.driverLabels(app))).create();
+ createWorkload(app);
+ Assertions.assertEquals(1, podDeleteOptions().size());
+
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+
+ // Every driver is looked at, so the one of the suspended attempt is
deleted as well, while the
+ // terminating one is not deleted again within its grace period
+ List<String> deletions = podDeleteOptions();
+ Assertions.assertEquals(1, deletions.size(), deletions.toString());
+
Assertions.assertTrue(deletions.get(0).contains("\"propagationPolicy\":\"Foreground\""));
+ Assertions.assertNull(getPod(NEXT_DRIVER));
+ Assertions.assertNotNull(getWorkload());
+ verifyNoInteractions(recorder, eventRecorder);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void podsWhichOutliveTheGracePeriodNoLongerHoldTheRelease(boolean suspend)
throws Exception {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
suspend);
+ // The wait for a pod ends this long after its own grace period ended, at
its deletionTimestamp
+ app.getSpec()
+ .getApplicationTolerations()
+ .getApplicationTimeoutConfig()
+ .setForceTerminationGracePeriodMillis(0L);
+ stubContext(app);
+ createDriver(app, FINALIZER);
+ createExecutor(app, FINALIZER);
+ createWorkload(app);
+ // Pods which are still terminating, e.g. on a lost node
+ kubernetesClient.resource(driver(app)).delete();
+ kubernetesClient.resource(executor(app)).delete();
+ Assertions.assertEquals(2, podDeleteOptions().size());
+
+ Assertions.assertEquals(
+ suspend ? SUSPEND_HOLD_PROGRESS :
ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ // The terminating driver is force deleted, so that it no longer keeps its
pod object
+ List<String> deletions = podDeleteOptions();
+ Assertions.assertEquals(1, deletions.size(), deletions.toString());
+
Assertions.assertTrue(deletions.get(0).contains("\"gracePeriodSeconds\":0"));
+ Assertions.assertNull(getWorkload());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void waitIsRequeuedAtTheNextPodDeadline(boolean suspend) {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
suspend);
+ app.getSpec()
+ .getApplicationTolerations()
+ .getApplicationTimeoutConfig()
+
.setForceTerminationGracePeriodMillis(Duration.ofMinutes(1).toMillis());
+ stubContext(app);
+ // A driver which is still terminating, e.g. on a lost node, and its
executor which is not
+ // deleted yet
+ createDriver(app, FINALIZER);
+ kubernetesClient.resource(driver(app)).delete();
+ createExecutor(app);
+ createWorkload(app);
+
+ ReconcileProgress progress = new AppSuspendStep().reconcile(mockContext,
recorder);
+
+ // The wait is requeued to end when the driver has been terminating for
the timeout, rather
+ // than after the default interval, since a driver on a lost node may send
no more events
+ Assertions.assertTrue(progress.isCompleted());
+ Duration requeueAfter = progress.getRequeueAfterDuration();
+ Assertions.assertTrue(
+ requeueAfter.compareTo(Duration.ofSeconds(30)) > 0
+ && requeueAfter.compareTo(Duration.ofMinutes(1)) <= 0,
+ requeueAfter::toString);
+ Assertions.assertNotNull(getWorkload());
+ verifyNoInteractions(recorder, eventRecorder);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void waitIsNotRequeuedAtPassedPodDeadlines(boolean suspend) {
+ SparkApplication app = buildApp(ApplicationStateSummary.Suspended,
suspend);
+ app.getSpec()
+ .getApplicationTolerations()
+ .getApplicationTimeoutConfig()
+ .setForceTerminationGracePeriodMillis(0L);
+ stubContext(app);
+ createDriver(app);
+ // An executor which is still terminating, e.g. on a lost node
+ createExecutor(app, FINALIZER);
+ kubernetesClient.resource(executor(app)).delete();
+
+ // Only the driver, whose deletion is observed by the pod informer, holds
the application, so
+ // the executor which no longer holds it does not requeue the wait at once
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+ Assertions.assertNull(getDriver());
+ }
+
+ @ParameterizedTest
+ @EnumSource(ResourceRetainPolicy.class)
+ void suspendedAppReleasesKueueWorkloadAfterPodsAreGone(ResourceRetainPolicy
policy) {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
true);
+ app.getSpec().getApplicationTolerations().setResourceRetainPolicy(policy);
+ stubContext(app);
+ createDriver(app);
+ createWorkload(app);
+
+ // The driver which is deleted is waited for
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+ Assertions.assertNull(getDriver());
+ Assertions.assertNotNull(getWorkload());
+
+ // The quota is given back whatever the retain policy is, and a resumed
application starts a
+ // new attempt, which does not reuse the resources of the suspended one
+ Assertions.assertEquals(
+ SUSPEND_HOLD_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+
+ Assertions.assertNull(getWorkload());
+ // Unlike an application held before its driver is requested, its status
says that it is
+ // suspended, so no SuspendHeld event is published
+ verifyNoInteractions(recorder, eventRecorder);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void workloadIsReleasedAfterQueueLabelIsRemoved(boolean suspend) {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
suspend);
+ createWorkload(app);
+ app.getMetadata().setLabels(Map.of());
+ stubContext(app);
+
+ new AppSuspendStep().reconcile(mockContext, recorder);
+
+ // The Workload admitted before the label was removed neither keeps the
quota while suspended,
+ // nor lends its admission to a resumed application whose label is added
back
+ Assertions.assertNull(getWorkload());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void kueueWorkloadIsNotAccessedWhenKueueIsDisabled(boolean suspend) {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
suspend);
+ createWorkload(app);
+ KubernetesClient client = spy(kubernetesClient);
+ stubContext(app, client);
+ TestUtils.setConfigKey(SparkOperatorConf.KUEUE_ENABLED, false);
+
+ Assertions.assertEquals(
+ suspend ? SUSPEND_HOLD_PROGRESS :
ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ // The queue label is ignored, and neither the suspended nor the resumed
application calls Kueue
+ verify(client, never()).resources(Workload.class);
+ Assertions.assertNotNull(getWorkload());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {403, 429, 500})
+ void persistentReleaseFailureIsReportedAndRetried(int code) {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
false);
+ createDriver(app);
+ KubernetesClient client = spy(kubernetesClient);
+ // Unlike a timeout or a 503, these do not clear on their own, e.g. an
admission webhook which
+ // is down keeps failing with 500, so they are reported
+ doThrow(new KubernetesClientException("failed", code, null))
+ .when(client)
+ .resource(any(Pod.class));
+ stubContext(app, client);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ EventRecord event = captureEvent();
+ Assertions.assertEquals(EventType.WARNING, event.type());
+ Assertions.assertEquals(EventUtils.REASON_SUSPEND_RELEASE_FAILED,
event.reason());
+ // It stays Suspended rather than resuming on what it still holds
+ verify(client, never()).resources(Workload.class);
+ verifyNoInteractions(recorder);
+ }
+
+ @Test
+ void transientReleaseFailureIsRetriedWithoutReport() {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
false);
+ createDriver(app);
+ KubernetesClient client = spy(kubernetesClient);
+ doThrow(new KubernetesClientException("Service Unavailable", 503, null))
+ .when(client)
+ .resource(any(Pod.class));
+ stubContext(app, client);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ verify(client, never()).resources(Workload.class);
+ verifyNoInteractions(recorder, eventRecorder);
+ }
+
+ @Test
+ void failedPodListIsRetriedWithoutReleasingKueueWorkload() {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
false);
+ createWorkload(app);
+ KubernetesClient client = spy(kubernetesClient);
+ doThrow(new KubernetesClientException("Service Unavailable", 503,
null)).when(client).pods();
+ stubContext(app, client);
+
+ // Whether the pods are gone is unknown, so the Workload is kept and
nothing is resumed
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ Assertions.assertNotNull(getWorkload());
+ verifyNoInteractions(recorder, eventRecorder);
+ }
+
+ @Test
+ void failedKueueWorkloadReleaseIsReportedAndRetried() {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
false);
+ stubContext(app);
+ createWorkload(app);
+ server
+ .expect()
+ .delete()
+
.withPath("/apis/kueue.x-k8s.io/v1beta2/namespaces/default/workloads/sparkapplication-app1")
+ .andReturn(403, null)
+ .once();
+
+ // The quota is not given back yet, so the application is neither held for
the hold interval
+ // nor resumed on the admission of the suspended attempt
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ Assertions.assertNotNull(getWorkload());
+ Assertions.assertEquals(EventUtils.REASON_SUSPEND_RELEASE_FAILED,
captureEvent().reason());
+ verifyNoInteractions(recorder);
+ }
+
+ @Test
+ void resumingSuspendedAppStartsNewAttemptOnceEverythingIsReleased() {
+ SparkApplication app = buildKueueApp(ApplicationStateSummary.Suspended,
false);
+ stubContext(app);
+ // Resumed before anything was released: the driver and the Workload are
still there
+ createDriver(app);
+ createWorkload(app);
+ long lastStateId = app.getStatus().getStateTransitionHistory().lastKey();
+ long attemptId =
app.getStatus().getCurrentAttemptSummary().getAttemptInfo().getId();
+
+ // The driver which is deleted is waited for, along with the Workload
+ Assertions.assertEquals(
+ WAITING_FOR_PODS_PROGRESS, new AppSuspendStep().reconcile(mockContext,
recorder));
+ Assertions.assertNull(getDriver());
+ Assertions.assertNotNull(getWorkload());
+ verifyNoInteractions(recorder);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+
+ ArgumentCaptor<ApplicationStatus> status =
ArgumentCaptor.forClass(ApplicationStatus.class);
+ verify(recorder).persistStatus(eq(mockContext), status.capture());
+ ApplicationState state = status.getValue().getCurrentState();
+ Assertions.assertEquals(ApplicationStateSummary.Submitted,
state.getCurrentStateSummary());
+ Assertions.assertEquals(Constants.APP_RESUMED_MESSAGE, state.getMessage());
+ // The history of the suspended attempt moves to the previous attempt,
like on a restart
+ Assertions.assertEquals(
+ Map.of(lastStateId + 1, state),
status.getValue().getStateTransitionHistory());
+ // A new attempt, see ApplicationStatus#resume
+ Assertions.assertEquals(
+ attemptId + 1,
status.getValue().getCurrentAttemptSummary().getAttemptInfo().getId());
+ // Whatever the Workload was admitted for, AppInitStep requests a new
admission from scratch
+ Assertions.assertNull(getWorkload());
+ }
+
+ @Test
+ void resumeWhichIsNotPersistedIsRetriedLater() {
+ SparkApplication app = buildApp(ApplicationStateSummary.Suspended, false);
+ stubContext(app);
+ when(recorder.persistStatus(any(), any())).thenReturn(false);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new AppSuspendStep().reconcile(mockContext, recorder));
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = ApplicationStateSummary.class,
+ mode = EnumSource.Mode.EXCLUDE,
+ names = {
+ "DriverRequested",
+ "DriverStarted",
+ "DriverReady",
+ "InitializedBelowThresholdExecutors",
+ "RunningHealthy",
+ "RunningWithPartialCapacity",
+ "RunningWithBelowThresholdExecutors",
+ "Suspended"
+ })
+ void otherStatesProceed(ApplicationStateSummary summary) {
+ // An initializing application is held by AppInitStep, and a stopping one
stops as usual
+ SparkApplication app = buildApp(summary, true);
+ stubContext(app);
+ createDriver(app);
+
+ Assertions.assertEquals(
+ ReconcileProgress.proceed(), new
AppSuspendStep().reconcile(mockContext, recorder));
+
+ verifyNoInteractions(recorder);
+ Assertions.assertNotNull(getDriver());
+ }
+
+ private void stubContext(SparkApplication app) {
+ stubContext(app, kubernetesClient);
+ }
+
+ private void stubContext(SparkApplication app, KubernetesClient client) {
+ when(recorder.appendNewStateAndPersist(any(), any())).thenReturn(true);
+ when(recorder.persistStatus(any(), any())).thenReturn(true);
+ when(mockContext.getResource()).thenReturn(app);
+ when(mockContext.getClient()).thenReturn(client);
+ when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+ }
+
+ private ApplicationState captureAppendedState() {
+ ArgumentCaptor<ApplicationState> captor =
ArgumentCaptor.forClass(ApplicationState.class);
+ verify(recorder).appendNewStateAndPersist(eq(mockContext),
captor.capture());
+ verify(recorder, never()).persistStatus(any(), any());
+ return captor.getValue();
+ }
+
+ private EventRecord captureEvent() {
+ ArgumentCaptor<EventRecord> event =
ArgumentCaptor.forClass(EventRecord.class);
+ verify(eventRecorder).record(event.capture());
+ return event.getValue();
+ }
+
+ /** Returns the delete options of the pod deletions which the mock server
received. */
+ private List<String> podDeleteOptions() throws InterruptedException {
+ List<String> deletions = new ArrayList<>();
+ RecordedRequest request = server.takeRequest(0, TimeUnit.SECONDS);
+ while (request != null) {
+ if ("DELETE".equals(request.getMethod()) &&
request.getPath().contains("/pods/")) {
+ deletions.add(request.getUtf8Body());
+ }
+ request = server.takeRequest(0, TimeUnit.SECONDS);
+ }
+ return deletions;
+ }
+
+ private Pod getDriver() {
+ return getPod(DRIVER);
+ }
+
+ private Pod getPod(String name) {
+ return kubernetesClient.pods().inNamespace("default").withName(name).get();
+ }
+
+ private Workload getWorkload() {
+ return kubernetesClient
+ .resources(Workload.class)
+ .inNamespace("default")
+ .withName("sparkapplication-app1")
+ .get();
+ }
+
+ private void createDriver(SparkApplication app, String... finalizers) {
+ kubernetesClient.resource(withFinalizers(driver(app),
finalizers)).create();
+ }
+
+ private void createExecutor(SparkApplication app, String... finalizers) {
+ kubernetesClient.resource(withFinalizers(executor(app),
finalizers)).create();
+ }
+
+ private void createWorkload(SparkApplication app) {
+
kubernetesClient.resource(KueueWorkloadFactory.buildWorkload(app)).create();
+ }
+
+ private static Pod driver(SparkApplication app) {
+ return pod(DRIVER, Utils.driverLabels(app));
+ }
+
+ private static Pod executor(SparkApplication app) {
+ return pod(EXECUTOR, Utils.executorLabels(app));
+ }
+
+ private static Pod pod(String name, Map<String, String> labels) {
+ return new PodBuilder()
+ .withNewMetadata()
+ .withName(name)
+ .withNamespace("default")
+ .withLabels(labels)
+ .endMetadata()
+ .build();
+ }
+
+ private static Pod withFinalizers(Pod pod, String... finalizers) {
+ return new
PodBuilder(pod).editMetadata().withFinalizers(finalizers).endMetadata().build();
+ }
+
+ private static SparkApplication buildApp(ApplicationStateSummary summary,
boolean suspend) {
+ SparkApplication app = new SparkApplication();
+ app.setMetadata(
+ new ObjectMetaBuilder()
+ .withName("app1")
+ .withNamespace("default")
+ .withUid("app-uid")
+ .build());
+ app.getSpec().setSuspend(suspend);
+ app.setStatus(app.getStatus().appendNewState(new ApplicationState(summary,
"")));
+ return app;
+ }
+
+ private static SparkApplication buildKueueApp(ApplicationStateSummary
summary, boolean suspend) {
+ SparkApplication app = buildApp(summary, suspend);
+ app.getMetadata().setLabels(Map.of(Constants.LABEL_QUEUE_NAME,
"test-queue"));
+ return app;
+ }
+}
diff --git a/tests/e2e/kueue/chainsaw-test.yaml
b/tests/e2e/kueue/chainsaw-test.yaml
index 3ae9a944..17981438 100644
--- a/tests/e2e/kueue/chainsaw-test.yaml
+++ b/tests/e2e/kueue/chainsaw-test.yaml
@@ -1052,6 +1052,76 @@ spec:
status:
currentState:
currentStateSummary: RunningHealthy
+ # As the event advises, suspending the application releases its driver and
executors, and then
+ # the quota which its evicted Workload keeps
+ - script:
+ timeout: 30s
+ content: |
+ kubectl patch sparkapplication
spark-job-kueue-executors-not-ready-test -n default --type merge -p
'{"spec":{"suspend":true}}'
+ - assert:
+ timeout: 60s
+ resource:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ metadata:
+ name: spark-job-kueue-executors-not-ready-test
+ namespace: default
+ status:
+ currentState:
+ currentStateSummary: Suspended
+ - error:
+ timeout: 120s
+ resource:
+ apiVersion: v1
+ kind: Pod
+ metadata:
+ namespace: default
+ labels:
+ spark.operator/spark-app-name:
spark-job-kueue-executors-not-ready-test
+ - error:
+ timeout: 120s
+ resource:
+ apiVersion: kueue.x-k8s.io/v1beta2
+ kind: Workload
+ metadata:
+ name: sparkapplication-spark-job-kueue-executors-not-ready-test
+ namespace: default
+ - assert:
+ timeout: 60s
+ resource:
+ apiVersion: kueue.x-k8s.io/v1beta2
+ kind: ClusterQueue
+ metadata:
+ name: ($CLUSTER_QUEUE)
+ status:
+ admittedWorkloads: 0
+ reservingWorkloads: 0
+ # Resuming queues it again with a new attempt and a new Workload, which
Kueue admits into the
+ # quota it gave back
+ - script:
+ timeout: 30s
+ content: |
+ kubectl patch sparkapplication
spark-job-kueue-executors-not-ready-test -n default --type merge -p
'{"spec":{"suspend":false}}'
+ - assert:
+ timeout: 5m
+ resource:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ metadata:
+ name: spark-job-kueue-executors-not-ready-test
+ namespace: default
+ status:
+ currentState:
+ currentStateSummary: RunningHealthy
+ - assert:
+ timeout: 60s
+ resource:
+ apiVersion: kueue.x-k8s.io/v1beta2
+ kind: ClusterQueue
+ metadata:
+ name: ($CLUSTER_QUEUE)
+ status:
+ admittedWorkloads: 1
catch:
- describe:
apiVersion: spark.apache.org/v1
diff --git a/tests/e2e/suspend/chainsaw-test.yaml
b/tests/e2e/suspend/chainsaw-test.yaml
index bff80456..e0601c85 100644
--- a/tests/e2e/suspend/chainsaw-test.yaml
+++ b/tests/e2e/suspend/chainsaw-test.yaml
@@ -78,6 +78,86 @@ spec:
timeout: 30s
content: |
kubectl delete sparkapplication spark-job-suspend-test -n default
--ignore-not-found=true
+ # Suspending a running application releases its driver and executors, and
resuming it starts a
+ # new attempt, which runs the application again from scratch.
+ - try:
+ - apply:
+ file: spark-example-running.yaml
+ - assert:
+ timeout: 120s
+ resource:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ metadata:
+ name: spark-job-suspend-running-test
+ namespace: default
+ status:
+ currentState:
+ currentStateSummary: RunningHealthy
+ - script:
+ timeout: 30s
+ content: |
+ kubectl patch sparkapplication spark-job-suspend-running-test -n
default --type merge -p '{"spec":{"suspend":true}}'
+ - assert:
+ timeout: 60s
+ resource:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ metadata:
+ name: spark-job-suspend-running-test
+ namespace: default
+ status:
+ currentState:
+ currentStateSummary: Suspended
+ - error:
+ timeout: 120s
+ resource:
+ apiVersion: v1
+ kind: Pod
+ metadata:
+ namespace: default
+ labels:
+ spark.operator/spark-app-name: spark-job-suspend-running-test
+ # The executor is made schedulable while the application is suspended, so
that the resumed
+ # attempt runs with the changed spec and completes.
+ - script:
+ timeout: 30s
+ content: |
+ kubectl patch sparkapplication spark-job-suspend-running-test -n
default --type merge -p
'{"spec":{"suspend":false,"sparkConf":{"spark.kubernetes.executor.node.selector.spark.apache.org/e2e-unschedulable":null}}}'
+ - assert:
+ timeout: 180s
+ resource:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ metadata:
+ name: spark-job-suspend-running-test
+ namespace: default
+ status:
+ currentState:
+ currentStateSummary: ResourceReleased
+ # The new attempt does not count as a restart, even with the
default restartPolicy
+ # Never
+ currentAttemptSummary:
+ attemptInfo:
+ id: 1
+ restartCounter: 0
+ stateTransitionHistory:
+ (contains(*.currentStateSummary, 'Succeeded')): true
+ previousAttemptSummary:
+ stateTransitionHistory:
+ (contains(*.currentStateSummary, 'Suspended')): true
+ catch:
+ - describe:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ namespace: default
+ - events:
+ namespace: default
+ finally:
+ - script:
+ timeout: 30s
+ content: |
+ kubectl delete sparkapplication spark-job-suspend-running-test -n
default --ignore-not-found=true
- try:
- apply:
bindings:
diff --git a/tests/e2e/suspend/spark-example-running.yaml
b/tests/e2e/suspend/spark-example-running.yaml
new file mode 100644
index 00000000..5bbfaab2
--- /dev/null
+++ b/tests/e2e/suspend/spark-example-running.yaml
@@ -0,0 +1,35 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+apiVersion: spark.apache.org/v1
+kind: SparkApplication
+metadata:
+ name: spark-job-suspend-running-test
+ namespace: default
+spec:
+ mainClass: "org.apache.spark.examples.SparkPi"
+ jars: "local:///opt/spark/examples/jars/spark-examples.jar"
+ sparkConf:
+ spark.executor.instances: "1"
+ # No node has this label, so the executor is never scheduled and the job
waits for it, while
+ # the application is RunningHealthy without executor thresholds. The test
removes it while the
+ # application is suspended, so that the resumed attempt completes.
+
spark.kubernetes.executor.node.selector.spark.apache.org/e2e-unschedulable:
"true"
+ spark.kubernetes.container.image: "apache/spark:{{SPARK_VERSION}}-scala"
+ spark.kubernetes.authenticate.driver.serviceAccountName: "spark"
+ runtimeVersions:
+ sparkVersion: "4.2.0"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
