You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/22 15:52:46 UTC
(spark-kubernetes-operator) branch main updated: [SPARK-60463] Release the resources of a `SparkCluster` on `SchedulingFailure` and end it as `ResourceReleased`
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 62350512 [SPARK-60463] Release the resources of a `SparkCluster` on `SchedulingFailure` and end it as `ResourceReleased`
62350512 is described below
commit 6235051264d7a2a72c5f936eb7473e54e70fa3a7
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Oct 22 08:52:38 2026 -0700
[SPARK-60463] Release the resources of a `SparkCluster` on `SchedulingFailure` and end it as `ResourceReleased`
### What changes were proposed in this pull request?
This PR aims to release the resources of a `SparkCluster` which fails with `SchedulingFailure`, and then to terminate it as `ResourceReleased` instead of `Failed`.
- A new `ClusterCleanUpStep` reconciles a cluster in `SchedulingFailure`, which `ClusterInitStep` records with the cause as before.
- It deletes the master and worker `StatefulSet`s and `Service`s, the `NetworkPolicy`, and the `HorizontalPodAutoscaler` and `PodDisruptionBudget` of the workers by name, with the name getters of `SparkClusterResourceSpec` (SPARK-60438), since the spec may not build, which may be the failure itself.
- Once the master and worker pods are gone, or are still terminating 5 minutes after their grace period ended, it deletes the Kueue `Workload` and moves the cluster to `ResourceReleased`, whose message points back to the preceding `SchedulingFailure` state for the cause.
- A failure of the release is retried with the default interval, and reported with the new `ClusterReleaseFailed` warning event unless it may clear on its own.
- `ClusterSuspendStep` shares the deletion by name (`deleteMasterAndWorkers`) and the wait for the pods (`waitForPods`) with it.
- A `Failed` cluster, which an earlier version leaves behind, is left as it is, so that an upgrade does not delete a master which may be in use.
### Why are the changes needed?
`ClusterInitStep` applies the `Service`s, the master and worker `StatefulSet`s, the `NetworkPolicy`, the `HorizontalPodAutoscaler` and the `PodDisruptionBudget` in order. When a request is rejected, e.g. with `403` by a `ResourceQuota` or an admission policy, or with `422`, the cluster moves to `SchedulingFailure` and then to `Failed`, which is never reconciled again. As a result,
- the master and workers which were requested before the rejection keep running,
- with `spark.kubernetes.operator.kueue.enabled`, the Kueue `Workload` keeps holding the quota, and
- neither `spec.suspend` nor a spec change releases them, since `ClusterSuspendStep` does not run in `Failed`. Only deleting the `SparkCluster` does.
The same happens when the spec of a queued cluster is changed to `minWorkers < maxWorkers` while its `Workload` waits for the admission, since `KueueWorkloadFactory.buildWorkload` throws `UnsupportedOperationException` for it. In contrast, the Kueue flavor conflict path of the same cluster (`KueueWorkloadUtils.holdForAdmission`) and `SparkApplication` (`AppCleanUpStep`) release their resources.
`ResourceReleased` is used as the final state because:
- It states that the resources are released. It has existed in `ClusterStateSummary` and the CRD since 0.1.0, so no CRD update is needed, and `isTerminated()` makes `ClusterTerminatedStep` end the reconciliation.
- A `SparkApplication` also moves from `SchedulingFailure` to `ResourceReleased` after releasing its resources.
- It tells a released cluster apart from a `Failed` cluster which an earlier version left behind with its resources.
- The cause stays in the preceding `SchedulingFailure` state and its warning event.
### Does this PR introduce _any_ user-facing change?
Yes. Compared with [1.0.0](https://github.com/apache/spark-kubernetes-operator/releases/tag/1.0.0) (2026-07-23), a `SparkCluster` which fails with `SchedulingFailure` releases the resources which it may have requested and ends in `ResourceReleased` instead of `Failed`. `docs/migration_guide.md` describes it, including that a `Failed` cluster which 1.0.0 left behind has to be deleted to release its resources.
The Kueue integration and the K8s events are not released yet, so releasing the `Workload` and the `ClusterReleaseFailed` event need no migration item. `architecture.md`, `configuration.md` and `spark_custom_resources.md` are updated accordingly.
### How was this patch tested?
Pass the CIs with the newly added test cases. The following test cases of `ClusterInitStepTest` reconcile a cluster with `SparkClusterReconciler`, and they fail without this fix.
- `rejectedRequestReleasesResourcesRequestedBeforeIt`: the worker `StatefulSet` is rejected after the master was requested.
- `rejectedRequestReleasesKueueWorkloadOncePodsAreGone`: the `PodDisruptionBudget` is rejected after the `Workload` was admitted.
- `unsupportedSpecOfQueuedClusterReleasesPendingKueueWorkload`: the spec of a queued cluster is changed to `minWorkers < maxWorkers`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #980 from dongjoon-hyun/SPARK-60463.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
docs/architecture.md | 20 +-
docs/configuration.md | 5 +-
docs/migration_guide.md | 14 +-
docs/spark_custom_resources.md | 35 ++-
.../org/apache/spark/k8s/operator/Constants.java | 10 +-
.../reconciler/SparkClusterReconciler.java | 5 +-
.../reconcilesteps/ClusterCleanUpStep.java | 124 ++++++++
.../reconciler/reconcilesteps/ClusterInitStep.java | 2 +
.../reconcilesteps/ClusterReconcileStep.java | 4 +-
.../reconcilesteps/ClusterSuspendStep.java | 139 +++++----
.../reconcilesteps/ClusterUnknownStateStep.java | 13 +-
.../spark/k8s/operator/utils/EventUtils.java | 6 +
.../reconciler/SparkClusterReconcilerTest.java | 41 +--
.../reconcilesteps/ClusterCleanUpStepTest.java | 319 +++++++++++++++++++++
.../reconcilesteps/ClusterInitStepTest.java | 231 +++++++++++++++
15 files changed, 861 insertions(+), 107 deletions(-)
diff --git a/docs/architecture.md b/docs/architecture.md
index 0b9fd4a0..3f2f0850 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -192,6 +192,9 @@ flowchart TD
reactivated. The resource then moves to `Submitted` and is queued again with a new `Workload`,
once `spec.suspend` is cleared if it is set. The `suspendReason` of the `Suspended` state,
`SpecSuspend` or `KueueEviction`, tells the two causes apart.
+ * When a `SparkCluster` fails with `SchedulingFailure`, `ClusterCleanUpStep` releases the master
+ and workers which were requested before the failure, and then the `Workload`, before the
+ cluster moves to `ResourceReleased`.
* When a `SparkCluster` is deleted, its `Workload` is garbage collected through its
`ownerReference`, like the other resources of the cluster.
* The Operator writes none of the `Workload` status which Kueue owns, such as the admission or the
@@ -320,7 +323,7 @@ stateDiagram-v2
Submitted --> RunningHealthy
Submitted --> SchedulingFailure
- SchedulingFailure --> Failed
+ SchedulingFailure --> ResourceReleased : Resources released
RunningHealthy --> Suspended : spec.suspend=true or Kueue eviction
Suspended --> Submitted : spec.suspend=false or after Kueue eviction
@@ -328,7 +331,7 @@ stateDiagram-v2
RunningHealthy --> [*]
Suspended --> [*]
- Failed --> [*]
+ ResourceReleased --> [*]
```
* Spark clusters are expected to be always running after submitted, until they are suspended by
@@ -339,8 +342,13 @@ stateDiagram-v2
* Likewise, a cluster stays in `Submitted` while it waits for the Kueue admission, or while it is
held by `spec.suspend` before it ever ran. See
[Kueue Admission and Workload Lifecycle](#kueue-admission-and-workload-lifecycle).
-* Apart from `spec.suspend` and a Kueue eviction, a cluster leaves `RunningHealthy`, `Suspended` or
- `Failed` only when its custom resource is deleted. At that point, the K8s resources created for the cluster are
+* Apart from `spec.suspend` and a Kueue eviction, a cluster leaves `RunningHealthy` or `Suspended`
+ only when its custom resource is deleted. At that point, the K8s resources created for the cluster are
garbage collected through their `ownerReference` to the `SparkCluster` custom resource.
-* A `Failed` cluster is not reconciled any further.
-* `ResourceReleased` exists in the API enum but is currently not used for clusters.
+* A cluster whose resources cannot be requested records the cause in `SchedulingFailure`, in which
+ `ClusterCleanUpStep` deletes by name what may have been requested, and releases the Kueue
+ `Workload` once the master and worker pods are gone. Then it moves to `ResourceReleased`, which
+ is not reconciled any further.
+* A cluster no longer enters `Failed`, which an earlier version entered after `SchedulingFailure`
+ without releasing the resources. Such a `Failed` cluster is not reconciled any further, so delete
+ it to release them.
diff --git a/docs/configuration.md b/docs/configuration.md
index ece6014a..c53932c1 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -96,7 +96,7 @@ replacing its `message`. A repeat carrying the same `message` may be paced by
| Type | Reason |
|---|---|
| `Warning` | `SparkApplication`: `SchedulingFailure`, `Failed`, `DriverEvicted`, `DriverStartTimedOut`, `DriverReadyTimedOut`, `ExecutorsStartTimedOut`, `RunningWithBelowThresholdExecutors`, `TerminatedWithoutReleaseResources` |
-| `Warning` | `SparkCluster`: `SchedulingFailure`, `Failed` |
+| `Warning` | `SparkCluster`: `SchedulingFailure` |
| `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
@@ -113,6 +113,7 @@ 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. |
+| `ClusterReleaseFailed` | Deleting the StatefulSets, Services, NetworkPolicy, HorizontalPodAutoscaler or PodDisruptionBudget, or the Kueue `Workload` of a `SparkCluster` which failed with `SchedulingFailure` 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, and the cluster stays in `SchedulingFailure` until it succeeds. |
| `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 it and queues it again, a `SparkApplication` with a new attempt, even if its pods are ready. Transport-level errors and conflicts are skipped. A `SparkCluster` retries them every 5 seconds and a persistent failur [...]
@@ -216,7 +217,7 @@ and `spark.kubernetes.executor.secretKeyRef.[EnvName]`. To keep the details of f
resource status and the operator log only, exclude their reasons, e.g. as follows.
```properties
-spark.kubernetes.operator.events.excludedReasons=SchedulingFailure,ReconcileError,CleanupError,StatusUpdateFailed,ClusterRequestFailed,SuspendReleaseFailed,Kueue.*Failed
+spark.kubernetes.operator.events.excludedReasons=SchedulingFailure,ReconcileError,CleanupError,StatusUpdateFailed,ClusterRequestFailed,ClusterReleaseFailed,SuspendReleaseFailed,Kueue.*Failed
```
## Metrics
diff --git a/docs/migration_guide.md b/docs/migration_guide.md
index e1e7a116..b60fb382 100644
--- a/docs/migration_guide.md
+++ b/docs/migration_guide.md
@@ -228,9 +228,17 @@ is recreated, e.g. by suspending and resuming it with [`spec.suspend`](spark_cus
server never answered, stays `Submitted` and is retried at the next reconciliation, instead of
failing with `SchedulingFailure` and then `Failed`. A request which the API server rejected
still fails the cluster.
-- Since 1.1.0, the message of the `Failed` state after `SchedulingFailure` is
- `Cluster failed, since requesting its resources failed. See the preceding SchedulingFailure state for the cause.`
- instead of `Cannot process cluster status.`.
+- Since 1.1.0, a cluster which fails with `SchedulingFailure`, e.g. since a `ResourceQuota` or an
+ admission webhook rejects one of its resources, deletes the resources which it may have requested
+ before the failure, i.e. the master and worker StatefulSets and Services, the NetworkPolicy, and
+ the HorizontalPodAutoscaler and PodDisruptionBudget of the workers. Then it moves to
+ `ResourceReleased` with the message
+ `Cluster failed, since requesting its resources failed, and they are released. See the preceding SchedulingFailure state for the cause.`
+ 1.0 moved it to `Failed` with the message `Cannot process cluster status.` instead, and kept
+ them until the `SparkCluster` was deleted, e.g. a master which was requested before its workers
+ were rejected kept running. A tool which waits for `Failed` has to wait for `ResourceReleased`
+ instead. A `Failed` cluster which 1.0 left behind is not released, so delete it to release its
+ resources.
### Docker Image
diff --git a/docs/spark_custom_resources.md b/docs/spark_custom_resources.md
index 28fdc654..e9d95fb3 100644
--- a/docs/spark_custom_resources.md
+++ b/docs/spark_custom_resources.md
@@ -736,7 +736,7 @@ relates it to the reconciliation.
templates. A missing request defaults to the limit, or else to 1 CPU and `SPARK_DAEMON_MEMORY`
plus overhead. A worker uses `SPARK_WORKER_CORES` for the CPU and adds `SPARK_WORKER_MEMORY` to
the memory when they are set. A `SparkCluster` keeps the quota until it is deleted, suspended or
- evicted.
+ evicted, or until it fails to request its master and workers, see [Spark Cluster](#spark-cluster).
Set a cpu and memory request or limit on the `worker` container, or `SPARK_WORKER_CORES` and
`SPARK_WORKER_MEMORY`: a worker with none of them advertises the whole node to its executors,
well above the default the `Workload` requests.
@@ -746,8 +746,9 @@ relates it to the reconciliation.
stay on the pods even if the `kueue.x-k8s.io/queue-name` label is removed after the driver (or
master) was created, since the `Workload` keeps the quota until it is released with the
resources. A node label that conflicts with the node selector of the pods fails the resource
- with `SchedulingFailure`, and deletes the `Workload` unless its driver or master is running
- already, whose pods still hold the quota it reserved. Only fixing the node selector or the flavor
+ with `SchedulingFailure`, and deletes the `Workload` right away unless its driver or master is
+ running already, whose pods still hold the quota it reserved until they are released. Only
+ fixing the node selector or the flavor
resolves it: a restarted attempt requests the quota again and hits the same conflict, so an
application which restarts on `SchedulingFailure` should bound the attempts with
[`maxRestartOnSchedulingFailure`](#granular-restart-control). `ResourceFlavor`s are
@@ -801,15 +802,16 @@ relates it to the reconciliation.
* 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 `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
+ without the access, a suspended `SparkApplication` or `SparkCluster` does not resume, a stopping
+ `SparkApplication` neither terminates nor restarts, and a `SparkCluster` in `SchedulingFailure`
+ does not terminate, 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 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
- owner has no running pods, e.g. of a `Failed` `SparkCluster`.
+ owner has no running pods.
* A `SparkCluster` with `minWorkers < maxWorkers` and pod template files set through
`spark.kubernetes.{driver,executor}.podTemplateFile` are not supported yet. Such a resource fails
with `SchedulingFailure` instead of being queued.
@@ -929,4 +931,19 @@ that Spark may overwrite certain fields.
The master and worker pods use `terminationGracePeriodSeconds: 0` by default to delete a
`SparkCluster` faster, unless their pod templates set it. A longer grace period delays stopping
-a [suspended](#suspend) cluster and releasing its [Kueue](#kueue) quota accordingly.
+a [suspended](#suspend) or failed cluster and releasing its [Kueue](#kueue) quota accordingly.
+
+A `SparkCluster` whose resources cannot be requested, e.g. since a `ResourceQuota` or an admission
+webhook rejects one of them, fails with `SchedulingFailure`, whose message gives the cause, while a
+request which may yet succeed, such as a timeout, a throttled request or an internal server error,
+is retried. Then the operator deletes by name whatever it may have requested for the cluster: the
+master and worker StatefulSets and Services, the NetworkPolicy, and the HorizontalPodAutoscaler and
+PodDisruptionBudget of the workers. Once the master and worker pods are gone, or are still
+terminating 5 minutes after their grace period ended, it releases the [Kueue](#kueue) `Workload` of
+the cluster as well, and the cluster moves to `ResourceReleased`, which is not reconciled any
+further. So delete and recreate the cluster to request its resources again. A failure to delete
+them or the `Workload`, other than one which may clear on its own, publishes the
+`ClusterReleaseFailed` [event](configuration.md#kubernetes-events) by default, and the cluster stays
+in `SchedulingFailure` until the deletion succeeds. The Services and the NetworkPolicy hold no
+quota, so they are deleted after the `Workload` is released, and a failure to delete them does not
+hold the quota.
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 c3b49562..308522a1 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
@@ -311,12 +311,12 @@ public final class Constants {
"Cluster is submitted again after its eviction by Kueue.";
/**
- * Message indicating that the cluster failed after SchedulingFailure, whose message tells the
- * cause.
+ * Message indicating that the resources of the cluster are released after SchedulingFailure,
+ * whose message tells the cause.
*/
- public static final String CLUSTER_FAILED_AFTER_SCHEDULING_FAILURE_MESSAGE =
- "Cluster failed, since requesting its resources failed. See the preceding "
- + "SchedulingFailure state for the cause.";
+ public static final String CLUSTER_RELEASED_AFTER_SCHEDULING_FAILURE_MESSAGE =
+ "Cluster failed, since requesting its resources failed, and they are released. See the "
+ + "preceding SchedulingFailure state for the cause.";
/** Message indicating that the cluster status cannot be processed. */
public static final String UNKNOWN_CLUSTER_STATE_MESSAGE = "Cannot process cluster status.";
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
index a35a9265..7bf0ebf7 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
@@ -192,8 +192,9 @@ public class SparkClusterReconciler implements Reconciler<SparkCluster>, Cleaner
case Submitted -> steps.add(new ClusterInitStep());
// Spark Cluster is supposed to run infinitely, until it is suspended by spec.suspend.
case RunningHealthy, Suspended -> steps.add(new ClusterSuspendStep());
- case SchedulingFailure, Failed, ResourceReleased ->
- steps.add(new ClusterUnknownStateStep());
+ // A cluster which failed to request its resources releases them and terminates.
+ case SchedulingFailure -> steps.add(new ClusterCleanUpStep());
+ case Failed, ResourceReleased -> steps.add(new ClusterUnknownStateStep());
}
return steps;
}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterCleanUpStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterCleanUpStep.java
new file mode 100644
index 00000000..13fde4c9
--- /dev/null
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterCleanUpStep.java
@@ -0,0 +1,124 @@
+/*
+ * 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.CLUSTER_RELEASED_AFTER_SCHEDULING_FAILURE_MESSAGE;
+import static org.apache.spark.k8s.operator.SparkClusterResourceSpec.getMasterServiceName;
+import static org.apache.spark.k8s.operator.SparkClusterResourceSpec.getWorkerNetworkPolicyName;
+import static org.apache.spark.k8s.operator.SparkClusterResourceSpec.getWorkerServiceName;
+import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.completeAndDefaultRequeue;
+import static org.apache.spark.k8s.operator.reconciler.ReconcileProgress.proceed;
+
+import java.util.List;
+import java.util.Optional;
+
+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.SparkCluster;
+import org.apache.spark.k8s.operator.context.SparkClusterContext;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
+import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import org.apache.spark.k8s.operator.status.ClusterState;
+import org.apache.spark.k8s.operator.status.ClusterStateSummary;
+import org.apache.spark.k8s.operator.utils.EventUtils;
+import org.apache.spark.k8s.operator.utils.ReconcilerUtils;
+import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
+
+/**
+ * Releases the resources of a cluster which failed with SchedulingFailure, and then terminates it
+ * with ResourceReleased. Nothing requests them again, so the master and workers which were
+ * requested before the failure would otherwise keep running, and its Kueue Workload would keep its
+ * quota, until the cluster is deleted.
+ */
+@Slf4j
+public final class ClusterCleanUpStep extends ClusterReconcileStep {
+ /**
+ * Releases the resources of a cluster in SchedulingFailure, which {@link ClusterInitStep} records
+ * with the cause before anything is released, so that the release, which lasts until the pods
+ * are gone, has a state to continue in. Like {@link ClusterSuspendStep}, everything is deleted by
+ * name, since the spec may not build, which may be the failure itself, and the resources may be
+ * left from before the cluster was suspended. Unlike a suspended cluster, the Services and the
+ * NetworkPolicy are deleted as well, since nothing applies them again.
+ *
+ * <p>The Workload is released only after the master and worker pods are gone, or are still
+ * terminating past their timeout, see {@link ClusterSuspendStep#waitForPods}. A pod stuck in
+ * terminating does not hold the cluster beyond that, since no pod of its name is created again.
+ * The Services and the NetworkPolicy hold no quota, so they are deleted only after the Workload
+ * is released, like the resources of an application in {@link AppCleanUpStep}. Otherwise a
+ * failure to delete them, e.g. by an operator which may not access them and so failed to apply
+ * them, would hold the quota forever. Everything here is idempotent, so a release that failed is
+ * simply retried. A failure which is not expected to clear on its own is reported, since
+ * SchedulingFailure reports the failure to request the resources, not the one to release them.
+ *
+ * @param context The SparkClusterContext for the cluster.
+ * @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
+ * @return The ReconcileProgress indicating the next step.
+ */
+ @Override
+ public ReconcileProgress reconcile(
+ SparkClusterContext context, SparkClusterStatusRecorder statusRecorder) {
+ SparkCluster cluster = context.getResource();
+ if (cluster.getStatus().getCurrentState().getCurrentStateSummary()
+ != ClusterStateSummary.SchedulingFailure) {
+ return proceed();
+ }
+ String namespace = cluster.getMetadata().getNamespace();
+ String name = cluster.getMetadata().getName();
+ KubernetesClient client = context.getClient();
+ try {
+ List<Pod> pods = ClusterSuspendStep.deleteMasterAndWorkers(client, cluster);
+ Optional<ReconcileProgress> waiting = ClusterSuspendStep.waitForPods(pods);
+ if (waiting.isPresent()) {
+ log.debug("Waiting for the pods of the failed cluster to be deleted.");
+ return waiting.get();
+ }
+ // A Workload admitted before the queue label was removed is released as well.
+ KueueWorkloadUtils.deleteWorkloadOf(client, cluster);
+ client.services().inNamespace(namespace).withName(getMasterServiceName(name)).delete();
+ client.services().inNamespace(namespace).withName(getWorkerServiceName(name)).delete();
+ client
+ .network()
+ .networkPolicies()
+ .inNamespace(namespace)
+ .withName(getWorkerNetworkPolicyName(name))
+ .delete();
+ } catch (KubernetesClientException e) {
+ log.warn("Failed to release the resources of the failed cluster, will retry.", e);
+ if (!ReconcilerUtils.isTransientError(e)) {
+ EventUtils.warn(
+ context.getEventRecorder(),
+ EventUtils.REASON_CLUSTER_RELEASE_FAILED,
+ "Failed to release the resources or the Kueue Workload of the cluster which failed "
+ + "to request them, will retry. "
+ + EventUtils.describe(e));
+ }
+ return completeAndDefaultRequeue();
+ }
+ return appendStateAndImmediateRequeue(
+ context,
+ statusRecorder,
+ new ClusterState(
+ ClusterStateSummary.ResourceReleased,
+ CLUSTER_RELEASED_AFTER_SCHEDULING_FAILURE_MESSAGE));
+ }
+}
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java
index c9f7347f..c867abaf 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStep.java
@@ -190,6 +190,8 @@ public final class ClusterInitStep extends ClusterReconcileStep {
/**
* Fails the cluster with SchedulingFailure for the given failure of requesting its resources.
+ * The resources requested before the failure, and its Kueue Workload, are released in
+ * SchedulingFailure, see {@link ClusterCleanUpStep}.
*
* @param context The SparkClusterContext for the cluster.
* @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterReconcileStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterReconcileStep.java
index 52846b96..7d64b0c4 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterReconcileStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterReconcileStep.java
@@ -29,8 +29,8 @@ import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
/** Basic reconcile step for cluster. */
public abstract sealed class ClusterReconcileStep
- permits ClusterInitStep, ClusterSuspendStep, ClusterTerminatedStep, ClusterUnknownStateStep,
- ClusterValidateStep {
+ permits ClusterCleanUpStep, ClusterInitStep, ClusterSuspendStep, ClusterTerminatedStep,
+ ClusterUnknownStateStep, ClusterValidateStep {
/**
* Reconciles a specific step for a Spark cluster.
*
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java
index 15871973..a1db1961 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterSuspendStep.java
@@ -264,62 +264,13 @@ public final class ClusterSuspendStep extends ClusterReconcileStep {
SparkClusterStatusRecorder statusRecorder,
boolean releaseWorkload) {
SparkCluster cluster = context.getResource();
- String namespace = cluster.getMetadata().getNamespace();
- String name = cluster.getMetadata().getName();
KubernetesClient client = context.getClient();
try {
- client
- .apps()
- .statefulSets()
- .inNamespace(namespace)
- .withName(getWorkerStatefulSetName(name))
- .delete();
- client
- .apps()
- .statefulSets()
- .inNamespace(namespace)
- .withName(getMasterStatefulSetName(name))
- .delete();
- client
- .autoscaling()
- .v2()
- .horizontalPodAutoscalers()
- .inNamespace(namespace)
- .withName(getWorkerHorizontalPodAutoscalerName(name))
- .delete();
- client
- .policy()
- .v1()
- .podDisruptionBudget()
- .inNamespace(namespace)
- .withName(getWorkerPodDisruptionBudgetName(name))
- .delete();
- List<Pod> pods =
- ReconcilerUtils.podsOf(
- client,
- namespace,
- LABEL_SPARK_CLUSTER_NAME,
- name,
- LABEL_SPARK_ROLE_MASTER_VALUE,
- LABEL_SPARK_ROLE_WORKER_VALUE)
- .list()
- .getItems();
- Instant now = Instant.now();
- Instant podReleaseDeadline =
- pods.stream()
- .map(ClusterSuspendStep::getPodReleaseDeadline)
- .max(Comparator.naturalOrder())
- .orElse(now);
- if (now.isBefore(podReleaseDeadline)) {
- // The deletion of each pod is observed by the pod informer, which reconciles again, while
- // the timeout of a pod which may send no more events is observed by the requeue.
+ List<Pod> pods = deleteMasterAndWorkers(client, cluster);
+ Optional<ReconcileProgress> waiting = waitForPods(pods);
+ if (waiting.isPresent()) {
log.debug("Waiting for the pods of the suspended cluster to be deleted.");
- ReconcileProgress defaultRequeue = completeAndDefaultRequeue();
- Duration remaining = Duration.between(now, podReleaseDeadline);
- return Optional.of(
- remaining.compareTo(defaultRequeue.getRequeueAfterDuration()) < 0
- ? completeAndRequeueAfter(remaining)
- : defaultRequeue);
+ return waiting;
}
if (releaseWorkload) {
// A Workload admitted before the queue label was removed is released as well.
@@ -366,7 +317,87 @@ public final class ClusterSuspendStep extends ClusterReconcileStep {
}
/**
- * Returns when a master or worker pod stops holding the Kueue Workload of a suspended cluster.
+ * Deletes the master and worker StatefulSets of the given cluster, and the
+ * HorizontalPodAutoscaler and the PodDisruptionBudget of its workers, by name, and lists its
+ * master and worker pods, see {@link #releaseResources}. {@link ClusterCleanUpStep} releases a
+ * cluster which failed to request its resources the same way.
+ *
+ * @param client The KubernetesClient.
+ * @param cluster The SparkCluster.
+ * @return The master and worker pods which remain, including the ones being deleted.
+ * @throws KubernetesClientException if a deletion or the listing fails.
+ */
+ static List<Pod> deleteMasterAndWorkers(
+ final KubernetesClient client, final SparkCluster cluster) {
+ String namespace = cluster.getMetadata().getNamespace();
+ String name = cluster.getMetadata().getName();
+ client
+ .apps()
+ .statefulSets()
+ .inNamespace(namespace)
+ .withName(getWorkerStatefulSetName(name))
+ .delete();
+ client
+ .apps()
+ .statefulSets()
+ .inNamespace(namespace)
+ .withName(getMasterStatefulSetName(name))
+ .delete();
+ client
+ .autoscaling()
+ .v2()
+ .horizontalPodAutoscalers()
+ .inNamespace(namespace)
+ .withName(getWorkerHorizontalPodAutoscalerName(name))
+ .delete();
+ client
+ .policy()
+ .v1()
+ .podDisruptionBudget()
+ .inNamespace(namespace)
+ .withName(getWorkerPodDisruptionBudgetName(name))
+ .delete();
+ return ReconcilerUtils.podsOf(
+ client,
+ namespace,
+ LABEL_SPARK_CLUSTER_NAME,
+ name,
+ LABEL_SPARK_ROLE_MASTER_VALUE,
+ LABEL_SPARK_ROLE_WORKER_VALUE)
+ .list()
+ .getItems();
+ }
+
+ /**
+ * Returns the progress to wait with while any of the given pods may still hold the Kueue
+ * Workload of its cluster, i.e. until {@link #POD_RELEASE_TIMEOUT} after its grace period ended,
+ * see {@link #releaseResources}.
+ *
+ * @param pods The master and worker pods which remain.
+ * @return The progress to wait with, or empty once none of the pods holds the Workload.
+ */
+ static Optional<ReconcileProgress> waitForPods(final List<Pod> pods) {
+ Instant now = Instant.now();
+ Instant podReleaseDeadline =
+ pods.stream()
+ .map(ClusterSuspendStep::getPodReleaseDeadline)
+ .max(Comparator.naturalOrder())
+ .orElse(now);
+ if (!now.isBefore(podReleaseDeadline)) {
+ return Optional.empty();
+ }
+ // The deletion of each pod is observed by the pod informer, which reconciles again, while the
+ // timeout of a pod which may send no more events is observed by the requeue.
+ ReconcileProgress defaultRequeue = completeAndDefaultRequeue();
+ Duration remaining = Duration.between(now, podReleaseDeadline);
+ return Optional.of(
+ remaining.compareTo(defaultRequeue.getRequeueAfterDuration()) < 0
+ ? completeAndRequeueAfter(remaining)
+ : defaultRequeue);
+ }
+
+ /**
+ * Returns when a master or worker pod stops holding the Kueue Workload of its cluster.
*
* @param pod The master or worker pod.
* @return {@link #POD_RELEASE_TIMEOUT} after its deletionTimestamp, when its grace period
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterUnknownStateStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterUnknownStateStep.java
index f9f060bd..57037523 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterUnknownStateStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterUnknownStateStep.java
@@ -29,7 +29,9 @@ import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
/** Abnormal state handler for clusters. */
public final class ClusterUnknownStateStep extends ClusterReconcileStep {
/**
- * Reconciles the cluster when it is in an unknown state, marking it as failed.
+ * Reconciles the cluster when it is in an unknown state, marking it as failed. A Failed cluster,
+ * e.g. one which an earlier version failed after SchedulingFailure without releasing its
+ * resources, is left as it is.
*
* @param context The SparkClusterContext for the cluster.
* @param statusRecorder The SparkClusterStatusRecorder for recording status updates.
@@ -45,13 +47,8 @@ public final class ClusterUnknownStateStep extends ClusterReconcileStep {
if (currentStateSummary == ClusterStateSummary.Failed) {
return ReconcileProgress.completeAndNoRequeue();
}
- // SchedulingFailure already reports the cause, so Failed points back to it rather than reading
- // as another, unexplained failure.
- String message =
- currentStateSummary == ClusterStateSummary.SchedulingFailure
- ? Constants.CLUSTER_FAILED_AFTER_SCHEDULING_FAILURE_MESSAGE
- : Constants.UNKNOWN_CLUSTER_STATE_MESSAGE;
- ClusterState state = new ClusterState(ClusterStateSummary.Failed, message);
+ ClusterState state =
+ new ClusterState(ClusterStateSummary.Failed, Constants.UNKNOWN_CLUSTER_STATE_MESSAGE);
statusRecorder.persistStatus(context, context.getResource().getStatus().appendNewState(state));
return ReconcileProgress.completeAndImmediateRequeue();
}
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 bb0ecd39..92926ab0 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
@@ -64,6 +64,12 @@ public final class EventUtils {
*/
public static final String REASON_CLUSTER_REQUEST_FAILED = "ClusterRequestFailed";
+ /**
+ * Reason for an event describing a failure to release the resources of a SparkCluster which
+ * failed with SchedulingFailure, which is retried.
+ */
+ public static final String REASON_CLUSTER_RELEASE_FAILED = "ClusterReleaseFailed";
+
/** Reason for an event describing that a resource is held by {@code spec.suspend}. */
public static final String REASON_SUSPEND_HELD = "SuspendHeld";
diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconcilerTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconcilerTest.java
index 41ca91b2..2ea8ccec 100644
--- a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconcilerTest.java
+++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconcilerTest.java
@@ -19,7 +19,6 @@
package org.apache.spark.k8s.operator.reconciler;
-import static org.apache.spark.k8s.operator.Constants.CLUSTER_FAILED_AFTER_SCHEDULING_FAILURE_MESSAGE;
import static org.apache.spark.k8s.operator.Constants.LABEL_SPARK_CLUSTER_NAME;
import static org.apache.spark.k8s.operator.utils.TestUtils.setConfigKey;
import static org.assertj.core.api.Assertions.assertThat;
@@ -33,8 +32,10 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
@@ -243,6 +244,21 @@ class SparkClusterReconcilerTest {
"ClusterUnknownStateStep", steps.get(2).getClass().getSimpleName());
}
+ @Test
+ void testGetReconcileStepsForSchedulingFailureCluster() {
+ // SchedulingFailure state should include ClusterCleanUpStep to release the resources
+ cluster.setStatus(cluster.getStatus().appendNewState(
+ new ClusterState(ClusterStateSummary.SchedulingFailure, "")));
+ List<ClusterReconcileStep> steps = reconciler.getReconcileSteps(cluster);
+ assertEquals(3, steps.size());
+ assertEquals(
+ "ClusterValidateStep", steps.get(0).getClass().getSimpleName());
+ assertEquals(
+ "ClusterTerminatedStep", steps.get(1).getClass().getSimpleName());
+ assertEquals(
+ "ClusterCleanUpStep", steps.get(2).getClass().getSimpleName());
+ }
+
@SuppressWarnings("PMD.UnusedLocalVariable")
@Test
void testReconcileFailedClusterDoesNotLoop() throws Exception {
@@ -253,28 +269,21 @@ class SparkClusterReconcilerTest {
when(mock.getResource()).thenReturn(cluster);
when(mock.getClient()).thenReturn(mockClient);
})) {
- cluster.setStatus(cluster.getStatus().appendNewState(
- new ClusterState(ClusterStateSummary.SchedulingFailure, "")));
-
- // SchedulingFailure moves to Failed once.
- reconciler.reconcile(cluster, mockContext);
- assertEquals(
- ClusterStateSummary.Failed,
- cluster.getStatus().getCurrentState().getCurrentStateSummary());
- // Failed points back to SchedulingFailure, which reports the cause.
- assertEquals(
- CLUSTER_FAILED_AFTER_SCHEDULING_FAILURE_MESSAGE,
- cluster.getStatus().getCurrentState().getMessage());
+ // An earlier version moved SchedulingFailure to Failed without releasing the resources.
+ cluster.setStatus(cluster.getStatus()
+ .appendNewState(new ClusterState(ClusterStateSummary.SchedulingFailure, ""))
+ .appendNewState(new ClusterState(ClusterStateSummary.Failed, "")));
int historySize = cluster.getStatus().getStateTransitionHistory().size();
- verify(mockRecorder, times(1)).persistStatus(any(), any());
- // A Failed cluster neither appends a new state nor requeues immediately.
+ // A Failed cluster neither appends a new state nor requeues immediately, and what it holds
+ // is left as it is until it is deleted.
for (int i = 0; i < 3; i++) {
var control = reconciler.reconcile(cluster, mockContext);
assertTrue(control.getScheduleDelay().isEmpty());
}
assertEquals(historySize, cluster.getStatus().getStateTransitionHistory().size());
- verify(mockRecorder, times(1)).persistStatus(any(), any());
+ verify(mockRecorder, never()).persistStatus(any(), any());
+ verifyNoInteractions(mockClient);
}
}
diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterCleanUpStepTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterCleanUpStepTest.java
new file mode 100644
index 00000000..5409e6cf
--- /dev/null
+++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterCleanUpStepTest.java
@@ -0,0 +1,319 @@
+/*
+ * 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.doReturn;
+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.times;
+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.List;
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+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.api.model.PodList;
+import io.fabric8.kubernetes.api.model.PodListBuilder;
+import io.fabric8.kubernetes.api.model.ServiceBuilder;
+import io.fabric8.kubernetes.api.model.apps.StatefulSetBuilder;
+import io.fabric8.kubernetes.api.model.autoscaling.v2.HorizontalPodAutoscalerBuilder;
+import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyBuilder;
+import io.fabric8.kubernetes.api.model.policy.v1.PodDisruptionBudgetBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
+import io.fabric8.kubernetes.client.dsl.MixedOperation;
+import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
+import io.fabric8.kubernetes.client.dsl.PodResource;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+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.CsvSource;
+import org.mockito.ArgumentCaptor;
+
+import org.apache.spark.k8s.operator.Constants;
+import org.apache.spark.k8s.operator.SparkCluster;
+import org.apache.spark.k8s.operator.config.SparkOperatorConf;
+import org.apache.spark.k8s.operator.context.SparkClusterContext;
+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.ClusterSpec;
+import org.apache.spark.k8s.operator.spec.ClusterTolerations;
+import org.apache.spark.k8s.operator.spec.RuntimeVersions;
+import org.apache.spark.k8s.operator.spec.WorkerInstanceConfig;
+import org.apache.spark.k8s.operator.status.ClusterState;
+import org.apache.spark.k8s.operator.status.ClusterStateSummary;
+import org.apache.spark.k8s.operator.utils.EventUtils;
+import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
+import org.apache.spark.k8s.operator.utils.TestUtils;
+
+@EnableKubernetesMockClient(crud = true)
+class ClusterCleanUpStepTest {
+ private KubernetesClient kubernetesClient;
+
+ private final SparkClusterContext mockContext = mock(SparkClusterContext.class);
+ private final SparkClusterStatusRecorder recorder = mock(SparkClusterStatusRecorder.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);
+ }
+
+ @Test
+ void failedClusterReleasesEverythingByNameAndTerminates() {
+ SparkCluster cluster = buildCluster(ClusterStateSummary.SchedulingFailure);
+ stubContext(cluster, kubernetesClient);
+ List<HasMetadata> resources =
+ List.of(
+ new StatefulSetBuilder().withMetadata(metadata("cluster1-master")).build(),
+ new StatefulSetBuilder().withMetadata(metadata("cluster1-worker")).build(),
+ new ServiceBuilder().withMetadata(metadata("cluster1-master-svc")).build(),
+ new ServiceBuilder().withMetadata(metadata("cluster1-worker-svc")).build(),
+ new NetworkPolicyBuilder().withMetadata(metadata("cluster1-worker")).build(),
+ new HorizontalPodAutoscalerBuilder()
+ .withMetadata(metadata("cluster1-worker-hpa"))
+ .build(),
+ new PodDisruptionBudgetBuilder().withMetadata(metadata("cluster1-worker-pdb")).build());
+ resources.forEach(resource -> kubernetesClient.resource(resource).create());
+ kubernetesClient.resource(KueueWorkloadFactory.buildWorkload(cluster)).create();
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new ClusterCleanUpStep().reconcile(mockContext, recorder));
+
+ // Unlike a suspended cluster, the Services and the NetworkPolicy are not applied again
+ for (HasMetadata resource : resources) {
+ Assertions.assertNull(
+ kubernetesClient.resource(resource).get(), resource.getMetadata().getName());
+ }
+ Assertions.assertNull(getWorkload());
+ // By name, so that neither a spec which cannot be built nor one which no longer asks for the
+ // HorizontalPodAutoscaler holds them
+ verify(mockContext, never()).getMasterStatefulSetSpec();
+ verify(mockContext, never()).getHorizontalPodAutoscalerSpec();
+ ClusterState state = captureAppendedState();
+ Assertions.assertEquals(ClusterStateSummary.ResourceReleased, state.getCurrentStateSummary());
+ Assertions.assertEquals(
+ Constants.CLUSTER_RELEASED_AFTER_SCHEDULING_FAILURE_MESSAGE, state.getMessage());
+ }
+
+ @Test
+ void stuckPodsDoNotHoldTheFailedCluster() {
+ SparkCluster cluster = buildCluster(ClusterStateSummary.SchedulingFailure);
+ KubernetesClient client = spy(kubernetesClient);
+ // Pods stuck in terminating, e.g. on a lost node, past their timeout
+ stubPodList(
+ client,
+ terminatingPod("cluster1-master-0", Constants.LABEL_SPARK_ROLE_MASTER_VALUE),
+ terminatingPod("cluster1-worker-0", Constants.LABEL_SPARK_ROLE_WORKER_VALUE));
+ stubContext(cluster, client);
+ kubernetesClient.resource(KueueWorkloadFactory.buildWorkload(cluster)).create();
+
+ // Unlike a suspended cluster, which may be resumed, nothing creates pods of the same names
+ // again, so the cluster terminates without waiting for them to be gone
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndImmediateRequeue(),
+ new ClusterCleanUpStep().reconcile(mockContext, recorder));
+
+ Assertions.assertNull(getWorkload());
+ Assertions.assertEquals(
+ ClusterStateSummary.ResourceReleased, captureAppendedState().getCurrentStateSummary());
+ }
+
+ @Test
+ void failedReleaseOfResourceWithoutQuotaDoesNotHoldKueueWorkload() {
+ SparkCluster cluster = buildCluster(ClusterStateSummary.SchedulingFailure);
+ KubernetesClient client = spy(kubernetesClient);
+ // e.g. an operator which may not access NetworkPolicies, which failed the cluster as well
+ doThrow(new KubernetesClientException("forbidden", 403, null)).when(client).network();
+ stubContext(cluster, client);
+ when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+ kubernetesClient.resource(KueueWorkloadFactory.buildWorkload(cluster)).create();
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new ClusterCleanUpStep().reconcile(mockContext, recorder));
+
+ // The NetworkPolicy holds no quota, so the Workload is released even though it is not deleted,
+ // while the cluster stays in SchedulingFailure and reports the failure
+ Assertions.assertNull(getWorkload());
+ verifyNoInteractions(recorder);
+ ArgumentCaptor<EventRecord> event = ArgumentCaptor.forClass(EventRecord.class);
+ verify(eventRecorder).record(event.capture());
+ Assertions.assertEquals(EventUtils.REASON_CLUSTER_RELEASE_FAILED, event.getValue().reason());
+ }
+
+ @ParameterizedTest
+ @CsvSource({"403, 1", "500, 1", "503, 0"})
+ void failedReleaseIsRetried(int code, int events) {
+ SparkCluster cluster = buildCluster(ClusterStateSummary.SchedulingFailure);
+ KubernetesClient mockClient = mock(KubernetesClient.class);
+ when(mockClient.apps()).thenThrow(new KubernetesClientException("failed", code, null));
+ stubContext(cluster, mockClient);
+ when(mockContext.getEventRecorder()).thenReturn(eventRecorder);
+
+ Assertions.assertEquals(
+ ReconcileProgress.completeAndDefaultRequeue(),
+ new ClusterCleanUpStep().reconcile(mockContext, recorder));
+
+ // It stays in SchedulingFailure, which reports the failure to request the resources, so the
+ // failure to release them is reported unless it may clear on its own
+ verify(mockClient, never()).resources(Workload.class);
+ verifyNoInteractions(recorder);
+ ArgumentCaptor<EventRecord> event = ArgumentCaptor.forClass(EventRecord.class);
+ verify(eventRecorder, times(events)).record(event.capture());
+ for (EventRecord record : event.getAllValues()) {
+ Assertions.assertEquals(EventType.WARNING, record.type());
+ Assertions.assertEquals(EventUtils.REASON_CLUSTER_RELEASE_FAILED, record.reason());
+ }
+ }
+
+ @Test
+ void otherStatesProceed() {
+ for (ClusterStateSummary summary : ClusterStateSummary.values()) {
+ if (summary == ClusterStateSummary.SchedulingFailure) {
+ continue;
+ }
+ SparkCluster cluster = buildCluster(summary);
+ KubernetesClient mockClient = mock(KubernetesClient.class);
+ stubContext(cluster, mockClient);
+
+ Assertions.assertEquals(
+ ReconcileProgress.proceed(), new ClusterCleanUpStep().reconcile(mockContext, recorder));
+
+ verifyNoInteractions(mockClient);
+ }
+ verifyNoInteractions(recorder);
+ }
+
+ private void stubContext(SparkCluster cluster, KubernetesClient client) {
+ when(recorder.appendNewStateAndPersist(any(), any())).thenReturn(true);
+ when(mockContext.getResource()).thenReturn(cluster);
+ when(mockContext.getClient()).thenReturn(client);
+ }
+
+ /**
+ * Stubs the LIST of the master and worker pods of cluster1 on the given spy to return the given
+ * pods, since the mock server does not evaluate set-based label selectors.
+ */
+ @SuppressWarnings("unchecked")
+ private static void stubPodList(KubernetesClient client, Pod... items) {
+ MixedOperation<Pod, PodList, PodResource> pods = mock(MixedOperation.class);
+ NonNamespaceOperation<Pod, PodList, PodResource> namespacedPods =
+ mock(NonNamespaceOperation.class);
+ FilterWatchListDeletable<Pod, PodList, PodResource> labeledPods =
+ mock(FilterWatchListDeletable.class);
+ FilterWatchListDeletable<Pod, PodList, PodResource> rolePods =
+ mock(FilterWatchListDeletable.class);
+ doReturn(pods).when(client).pods();
+ when(pods.inNamespace("default")).thenReturn(namespacedPods);
+ when(namespacedPods.withLabel(Constants.LABEL_SPARK_CLUSTER_NAME, "cluster1"))
+ .thenReturn(labeledPods);
+ when(labeledPods.withLabelIn(
+ Constants.LABEL_SPARK_ROLE_NAME,
+ Constants.LABEL_SPARK_ROLE_MASTER_VALUE,
+ Constants.LABEL_SPARK_ROLE_WORKER_VALUE))
+ .thenReturn(rolePods);
+ when(rolePods.list()).thenReturn(new PodListBuilder().addToItems(items).build());
+ }
+
+ private ClusterState captureAppendedState() {
+ ArgumentCaptor<ClusterState> captor = ArgumentCaptor.forClass(ClusterState.class);
+ verify(recorder).appendNewStateAndPersist(eq(mockContext), captor.capture());
+ verify(recorder, never()).persistStatus(any(), any());
+ return captor.getValue();
+ }
+
+ private Workload getWorkload() {
+ return kubernetesClient
+ .resources(Workload.class)
+ .inNamespace("default")
+ .withName("sparkcluster-cluster1")
+ .get();
+ }
+
+ private static SparkCluster buildCluster(ClusterStateSummary summary) {
+ SparkCluster cluster = new SparkCluster();
+ cluster.setMetadata(
+ new ObjectMetaBuilder()
+ .withName("cluster1")
+ .withNamespace("default")
+ .withUid("cluster-uid")
+ .withLabels(Map.of(Constants.LABEL_QUEUE_NAME, "cluster-queue"))
+ .build());
+ cluster.setSpec(
+ ClusterSpec.builder()
+ .runtimeVersions(RuntimeVersions.builder().sparkVersion("4.2.0").build())
+ .clusterTolerations(
+ ClusterTolerations.builder()
+ .instanceConfig(
+ WorkerInstanceConfig.builder()
+ .initWorkers(1)
+ .minWorkers(1)
+ .maxWorkers(1)
+ .build())
+ .build())
+ .build());
+ cluster.setStatus(cluster.getStatus().appendNewState(new ClusterState(summary, "")));
+ return cluster;
+ }
+
+ private static ObjectMeta metadata(String name) {
+ return new ObjectMetaBuilder().withName(name).withNamespace("default").build();
+ }
+
+ /** Builds a pod which is still terminating 6 minutes past its deletion. */
+ private static Pod terminatingPod(String name, String role) {
+ return new PodBuilder()
+ .withNewMetadata()
+ .withName(name)
+ .withNamespace("default")
+ .withDeletionTimestamp(Instant.now().minus(Duration.ofMinutes(6)).toString())
+ .addToLabels(Constants.LABEL_SPARK_CLUSTER_NAME, "cluster1")
+ .addToLabels(Constants.LABEL_SPARK_ROLE_NAME, role)
+ .endMetadata()
+ .build();
+ }
+}
diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java
index c32511c8..64baaa32 100644
--- a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java
+++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/ClusterInitStepTest.java
@@ -20,6 +20,8 @@
package org.apache.spark.k8s.operator.reconciler.reconcilesteps;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.inOrder;
@@ -31,13 +33,19 @@ import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
import java.time.Duration;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import io.fabric8.kubernetes.api.model.ConditionBuilder;
+import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.KubernetesResourceList;
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.api.model.PodList;
+import io.fabric8.kubernetes.api.model.PodListBuilder;
import io.fabric8.kubernetes.api.model.PodSpec;
import io.fabric8.kubernetes.api.model.Service;
import io.fabric8.kubernetes.api.model.ServiceBuilder;
@@ -51,8 +59,10 @@ import io.fabric8.kubernetes.api.model.policy.v1.PodDisruptionBudget;
import io.fabric8.kubernetes.api.model.scheduling.v1.PriorityClassList;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
import io.fabric8.kubernetes.client.dsl.MixedOperation;
import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
+import io.fabric8.kubernetes.client.dsl.PodResource;
import io.fabric8.kubernetes.client.dsl.Resource;
import io.fabric8.kubernetes.client.dsl.RollableScalableResource;
import io.fabric8.kubernetes.client.dsl.ServerSideApplicable;
@@ -71,6 +81,7 @@ import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
+import org.mockito.stubbing.Answer;
import org.apache.spark.k8s.operator.Constants;
import org.apache.spark.k8s.operator.SparkCluster;
@@ -86,7 +97,9 @@ import org.apache.spark.k8s.operator.kueue.v1beta2.ResourceFlavor;
import org.apache.spark.k8s.operator.kueue.v1beta2.ResourceFlavorSpec;
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.ReconcileProgress;
+import org.apache.spark.k8s.operator.reconciler.SparkClusterReconciler;
import org.apache.spark.k8s.operator.spec.ClusterSpec;
import org.apache.spark.k8s.operator.spec.ClusterTolerations;
import org.apache.spark.k8s.operator.spec.MasterSpec;
@@ -310,6 +323,224 @@ class ClusterInitStepTest {
verifyNoInteractions(eventRecorder);
}
+ @Test
+ @SuppressWarnings("unchecked")
+ void rejectedRequestReleasesResourcesRequestedBeforeIt() throws Exception {
+ SparkCluster cluster = buildKueueCluster();
+ cluster.getMetadata().setLabels(Map.of());
+ KubernetesClient mockClient = mock(KubernetesClient.class, RETURNS_DEEP_STUBS);
+ when(mockClient.resource(isA(StatefulSet.class)).get()).thenReturn(null);
+ // e.g. a ResourceQuota on the number of StatefulSets rejects the worker after the master
+ RollableScalableResource<StatefulSet> rejected =
+ mock(RollableScalableResource.class, RETURNS_DEEP_STUBS);
+ when(rejected.forceConflicts().serverSideApply())
+ .thenThrow(new KubernetesClientException("exceeded quota", 403, null));
+ when(mockClient
+ .apps()
+ .statefulSets()
+ .resource(
+ argThat((StatefulSet s) -> "cluster1-worker".equals(s.getMetadata().getName()))))
+ .thenReturn(rejected);
+ var statefulSets =
+ inDefaultNamespace(mockClient.apps().statefulSets(), RollableScalableResource.class);
+ var services = inDefaultNamespace(mockClient.services(), ServiceResource.class);
+ var networkPolicies =
+ inDefaultNamespace(mockClient.network().networkPolicies(), Resource.class);
+ inDefaultNamespace(mockClient.autoscaling().v2().horizontalPodAutoscalers(), Resource.class);
+ inDefaultNamespace(mockClient.policy().v1().podDisruptionBudget(), Resource.class);
+ stubWorkloadRead(mockClient, null);
+ stubPodLists(mockClient, new PodList());
+ SparkClusterReconciler reconciler = reconcilerOf(cluster);
+ Context<SparkCluster> josdkContext = josdkContextOf(mockClient);
+
+ reconciler.reconcile(cluster, josdkContext);
+ Assertions.assertEquals(
+ ClusterStateSummary.SchedulingFailure,
+ cluster.getStatus().getCurrentState().getCurrentStateSummary());
+
+ // The master and the Services requested before the rejection are released rather than kept
+ // until the cluster is deleted, and the cluster terminates
+ reconciler.reconcile(cluster, josdkContext);
+ verify(statefulSets.withName("cluster1-master")).delete();
+ verify(statefulSets.withName("cluster1-worker")).delete();
+ verify(services.withName("cluster1-master-svc")).delete();
+ verify(services.withName("cluster1-worker-svc")).delete();
+ verify(networkPolicies.withName("cluster1-worker")).delete();
+ Assertions.assertEquals(
+ ClusterStateSummary.ResourceReleased,
+ cluster.getStatus().getCurrentState().getCurrentStateSummary());
+ Assertions.assertTrue(
+ reconciler.reconcile(cluster, josdkContext).getScheduleDelay().isEmpty());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void rejectedRequestReleasesKueueWorkloadOncePodsAreGone() throws Exception {
+ SparkCluster cluster = buildKueueCluster();
+ KubernetesClient mockClient = mock(KubernetesClient.class, RETURNS_DEEP_STUBS);
+ // Mockito cannot deep-stub the generic list, so it is stubbed without any priority class
+ when(mockClient.scheduling().v1().priorityClasses().list())
+ .thenReturn(new PriorityClassList());
+ when(mockClient.resource(isA(StatefulSet.class)).get()).thenReturn(null);
+ when(mockClient.resource(isA(Workload.class)).get()).thenReturn(admittedWorkload(cluster));
+ // e.g. an admission policy rejects the PodDisruptionBudget after the master and workers
+ Resource<PodDisruptionBudget> rejected = mock(Resource.class, RETURNS_DEEP_STUBS);
+ when(rejected.forceConflicts().serverSideApply())
+ .thenThrow(new KubernetesClientException("denied by policy", 403, null));
+ when(mockClient.policy().v1().podDisruptionBudget().resource(any(PodDisruptionBudget.class)))
+ .thenReturn(rejected);
+ var statefulSets =
+ inDefaultNamespace(mockClient.apps().statefulSets(), RollableScalableResource.class);
+ inDefaultNamespace(mockClient.services(), ServiceResource.class);
+ inDefaultNamespace(mockClient.network().networkPolicies(), Resource.class);
+ inDefaultNamespace(mockClient.autoscaling().v2().horizontalPodAutoscalers(), Resource.class);
+ inDefaultNamespace(mockClient.policy().v1().podDisruptionBudget(), Resource.class);
+ Resource<Workload> workload = stubWorkloadRead(mockClient, admittedWorkload(cluster));
+ // The master and worker pods remain until they are deleted with their StatefulSets
+ stubPodLists(
+ mockClient,
+ new PodListBuilder()
+ .addToItems(
+ pod("cluster1-master-0", Constants.LABEL_SPARK_ROLE_MASTER_VALUE),
+ pod("cluster1-worker-0", Constants.LABEL_SPARK_ROLE_WORKER_VALUE))
+ .build(),
+ new PodList());
+ SparkClusterReconciler reconciler = reconcilerOf(cluster);
+ Context<SparkCluster> josdkContext = josdkContextOf(mockClient);
+
+ reconciler.reconcile(cluster, josdkContext);
+ Assertions.assertEquals(
+ ClusterStateSummary.SchedulingFailure,
+ cluster.getStatus().getCurrentState().getCurrentStateSummary());
+
+ // The admitted Workload is kept while the pods still occupy its quota
+ reconciler.reconcile(cluster, josdkContext);
+ verify(statefulSets.withName("cluster1-master")).delete();
+ verify(statefulSets.withName("cluster1-worker")).delete();
+ verify(workload, never()).delete();
+ Assertions.assertEquals(
+ ClusterStateSummary.SchedulingFailure,
+ cluster.getStatus().getCurrentState().getCurrentStateSummary());
+
+ // and released once they are gone, rather than holding the quota until the cluster is deleted
+ reconciler.reconcile(cluster, josdkContext);
+ reconciler.reconcile(cluster, josdkContext);
+ verify(workload).delete();
+ Assertions.assertEquals(
+ ClusterStateSummary.ResourceReleased,
+ cluster.getStatus().getCurrentState().getCurrentStateSummary());
+ }
+
+ @Test
+ void unsupportedSpecOfQueuedClusterReleasesPendingKueueWorkload() throws Exception {
+ SparkCluster cluster = buildKueueCluster();
+ SparkClusterReconciler reconciler = reconcilerOf(cluster);
+ Context<SparkCluster> josdkContext = josdkContextOf(kubernetesClient);
+
+ reconciler.reconcile(cluster, josdkContext);
+ Assertions.assertNotNull(getWorkload());
+
+ // The Workload cannot be built for minWorkers < maxWorkers, which Kueue does not support yet
+ cluster.getSpec().getClusterTolerations().getInstanceConfig().setMaxWorkers(2);
+ reconciler.reconcile(cluster, josdkContext);
+ Assertions.assertEquals(
+ ClusterStateSummary.SchedulingFailure,
+ cluster.getStatus().getCurrentState().getCurrentStateSummary());
+
+ // The Workload queued for the earlier spec is not admitted later into quota which nothing uses
+ reconciler.reconcile(cluster, josdkContext);
+ reconciler.reconcile(cluster, josdkContext);
+ Assertions.assertNull(getWorkload());
+ Assertions.assertEquals(
+ ClusterStateSummary.ResourceReleased,
+ cluster.getStatus().getCurrentState().getCurrentStateSummary());
+ }
+
+ /**
+ * Builds a reconciler which reconciles the given cluster like the operator, so that the steps
+ * after a failed request run as well. The status is applied to the cluster as it is persisted.
+ */
+ @SuppressWarnings("unchecked")
+ private static SparkClusterReconciler reconcilerOf(SparkCluster cluster) {
+ SparkClusterStatusRecorder recorder = mock(SparkClusterStatusRecorder.class);
+ Answer<Boolean> persist =
+ invocation -> {
+ cluster.setStatus(invocation.getArgument(1));
+ return true;
+ };
+ when(recorder.persistStatus(any(), any())).thenAnswer(persist);
+ when(recorder.persistStatus(any(), any(), any())).thenAnswer(persist);
+ when(recorder.appendNewStateAndPersist(any(), any()))
+ .thenAnswer(
+ invocation -> {
+ cluster.setStatus(cluster.getStatus().appendNewState(invocation.getArgument(1)));
+ return true;
+ });
+ return new SparkClusterReconciler(
+ new SparkClusterSubmissionWorker(), recorder, mock(SentinelManager.class));
+ }
+
+ @SuppressWarnings("unchecked")
+ private Context<SparkCluster> josdkContextOf(KubernetesClient client) {
+ Context<SparkCluster> josdkContext = mock(Context.class);
+ when(josdkContext.getClient()).thenReturn(client);
+ when(josdkContext.eventRecorder()).thenReturn(eventRecorder);
+ return josdkContext;
+ }
+
+ /**
+ * Stubs the given operation of a deep-stubbed client in the default namespace, whose generic
+ * types Mockito cannot resolve, with one resource of the given class per name, so that their
+ * deletions can be verified.
+ */
+ @SuppressWarnings("unchecked")
+ private static <T extends HasMetadata, L, R extends Resource<T>>
+ NonNamespaceOperation<T, L, R> inDefaultNamespace(
+ MixedOperation<T, L, R> operation, Class<?> resourceClass) {
+ NonNamespaceOperation<T, L, R> namespaced = mock(NonNamespaceOperation.class);
+ Map<String, R> resources = new HashMap<>();
+ when(namespaced.withName(anyString()))
+ .thenAnswer(
+ invocation ->
+ resources.computeIfAbsent(
+ invocation.getArgument(0), name -> (R) mock(resourceClass)));
+ when(operation.inNamespace("default")).thenReturn(namespaced);
+ return namespaced;
+ }
+
+ /** Stubs the LIST of the master and worker pods of cluster1 to return the given lists in turn. */
+ @SuppressWarnings("unchecked")
+ private static void stubPodLists(KubernetesClient client, PodList first, PodList... rest) {
+ MixedOperation<Pod, PodList, PodResource> pods = mock(MixedOperation.class);
+ NonNamespaceOperation<Pod, PodList, PodResource> namespacedPods =
+ mock(NonNamespaceOperation.class);
+ FilterWatchListDeletable<Pod, PodList, PodResource> labeledPods =
+ mock(FilterWatchListDeletable.class);
+ FilterWatchListDeletable<Pod, PodList, PodResource> rolePods =
+ mock(FilterWatchListDeletable.class);
+ when(client.pods()).thenReturn(pods);
+ when(pods.inNamespace("default")).thenReturn(namespacedPods);
+ when(namespacedPods.withLabel(Constants.LABEL_SPARK_CLUSTER_NAME, "cluster1"))
+ .thenReturn(labeledPods);
+ when(labeledPods.withLabelIn(
+ Constants.LABEL_SPARK_ROLE_NAME,
+ Constants.LABEL_SPARK_ROLE_MASTER_VALUE,
+ Constants.LABEL_SPARK_ROLE_WORKER_VALUE))
+ .thenReturn(rolePods);
+ when(rolePods.list()).thenReturn(first, rest);
+ }
+
+ private static Pod pod(String name, String role) {
+ return new PodBuilder()
+ .withNewMetadata()
+ .withName(name)
+ .withNamespace("default")
+ .addToLabels(Constants.LABEL_SPARK_CLUSTER_NAME, "cluster1")
+ .addToLabels(Constants.LABEL_SPARK_ROLE_NAME, role)
+ .endMetadata()
+ .build();
+ }
+
@Test
void resumedClusterSuspendedAgainGoesBackToSuspended() {
ClusterInitStep clusterInitStep = new ClusterInitStep();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org