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 a556c358 [SPARK-59866] Support K8s events for Kueue eviction of
running `SparkApplication`
a556c358 is described below
commit a556c358af01c9b610d7371c679c05b63e8ec8a4
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Sep 29 08:45:31 2026 -0700
[SPARK-59866] Support K8s events for Kueue eviction of running
`SparkApplication`
### What changes were proposed in this pull request?
This PR aims to publish the existing `KueueEvictionIgnored` warning event
when Kueue evicts or deactivates the `Workload` of a `SparkApplication` whose
driver is requested, like a running `SparkCluster` does for a
`PodsReadyTimeout`.
- A new `AppKueueEvictionStep` reports it from `DriverRequested` to
`RunningWithBelowThresholdExecutors`. It does not stop or requeue the
application.
- The message tells whether the `Workload` still holds its quota (evicted)
or not (deactivated).
- Like `KueueAdmissionPending`, the event is republished while it lasts,
subject to `minIntervalSeconds`.
### Why are the changes needed?
Currently, an eviction or a deactivation after the driver is requested is
neither handled nor reported, while the driver and executors keep running.
Kueue's own events are on the `Workload`, so `kubectl describe
sparkapplication` shows nothing, although a preempting workload may keep
waiting for the quota.
### Does this PR introduce _any_ user-facing change?
No for the released versions, because the Kueue integration is not released
yet in
[1.0.0](https://github.com/apache/spark-kubernetes-operator/releases/tag/1.0.0)
(2026-07-23).
### How was this patch tested?
Pass the CIs with the newly added test cases.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #906 from dongjoon-hyun/SPARK-59866.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
docs/configuration.md | 2 +-
docs/spark_custom_resources.md | 14 +-
.../operator/reconciler/SparkAppReconciler.java | 3 +
.../reconcilesteps/AppKueueEvictionStep.java | 84 ++++++++++++
.../reconcilesteps/AppReconcileStep.java | 4 +-
.../spark/k8s/operator/utils/EventUtils.java | 4 +-
.../reconciler/SparkAppReconcilerTest.java | 22 +++
.../reconcilesteps/AppKueueEvictionStepTest.java | 149 +++++++++++++++++++++
tests/e2e/kueue/chainsaw-test.yaml | 96 +++++++++++++
.../kueue/spark-example-executors-not-ready.yaml | 37 +++++
10 files changed, 405 insertions(+), 10 deletions(-)
diff --git a/docs/configuration.md b/docs/configuration.md
index 31731afb..40795590 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -96,7 +96,7 @@ In addition, the operator publishes the following `Warning`
events.
| `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`. Transport-level errors and conflicts are
skipped. A `SparkCluster` retries them every 5 seconds and a persistent failure
with the default interval, while a `SparkApplication` retries any failure with
its next reconciliation, so [...]
| `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 the `Workload` of a running
`SparkCluster` by a `PodsReadyTimeout`, which the operator does not act on, so
the master and workers keep running and holding its quota. It is republished on
every reconciliation while the eviction 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 not stopped,
or evicted the `Workload` of a running `SparkCluster` by a `PodsReadyTimeout`.
The operator does not act on it, so the driver and executors (or master and
workers) 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, [...]
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 7a34661e..fe296f4b 100644
--- a/docs/spark_custom_resources.md
+++ b/docs/spark_custom_resources.md
@@ -770,11 +770,15 @@ spec:
workers are ready, so that `blockAdmission` admits the next workload.
* Preemption is not honored yet after the driver of a `SparkApplication` is
requested, and neither
is a `PodsReadyTimeout`, e.g. of an application which runs with fewer
executors than
- `spark.executor.instances`. The operator checks the admission only before it
creates the
- resources, so after a later eviction the driver and executors keep running.
An evicted
- `Workload` keeps its Kueue quota meanwhile, while a deactivated one no
longer counts against it.
- 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.
+ `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 when enabled. 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
+ application whose pods are not all ready holds back every other workload
until they are ready
+ or its quota is released as described above.
## Spark Cluster
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 4a9ef3bf..8cf56fcd 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
@@ -56,6 +56,7 @@ import
org.apache.spark.k8s.operator.reconciler.observers.AppDriverStartObserver
import
org.apache.spark.k8s.operator.reconciler.observers.AppDriverTimeoutObserver;
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppCleanUpStep;
import org.apache.spark.k8s.operator.reconciler.reconcilesteps.AppInitStep;
+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.AppRunningStep;
@@ -207,6 +208,7 @@ public class SparkAppReconciler implements
Reconciler<SparkApplication>, Cleaner
switch (app.getStatus().getCurrentState().getCurrentStateSummary()) {
case Submitted, ScheduledToRestart -> steps.add(new AppInitStep());
case DriverRequested, DriverStarted -> {
+ steps.add(new AppKueueEvictionStep());
steps.add(
new AppResourceObserveStep(
List.of(new AppDriverStartObserver(), new
AppDriverReadyObserver())));
@@ -218,6 +220,7 @@ public class SparkAppReconciler implements
Reconciler<SparkApplication>, Cleaner
RunningHealthy,
RunningWithPartialCapacity,
RunningWithBelowThresholdExecutors -> {
+ steps.add(new AppKueueEvictionStep());
steps.add(new AppRunningStep());
steps.add(new AppResourceObserveStep(List.of(new
AppDriverRunningObserver())));
steps.add(new AppResourceObserveStep(List.of(new
AppDriverTimeoutObserver())));
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
new file mode 100644
index 00000000..cbeb760e
--- /dev/null
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStep.java
@@ -0,0 +1,84 @@
+/*
+ * 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.reconciler.ReconcileProgress.proceed;
+
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.Condition;
+
+import org.apache.spark.k8s.operator.context.SparkAppContext;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
+import org.apache.spark.k8s.operator.kueue.v1beta2.Workload;
+import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import org.apache.spark.k8s.operator.utils.EventUtils;
+import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
+
+/**
+ * Reports an eviction or a deactivation of the Kueue Workload of an
application whose driver is
+ * requested, which the operator does not act on yet. Unlike before the driver
is requested, see
+ * {@link AppInitStep}, the driver and executors keep running. The status does
not show it, so the
+ * event is published on every reconciliation while it lasts, like the pending
event of a queued
+ * application, and its message stays the same meanwhile, so that {@link
+ *
org.apache.spark.k8s.operator.config.SparkOperatorConf#KUBERNETES_EVENTS_MIN_INTERVAL_SECONDS}
+ * paces the repeats.
+ */
+public final class AppKueueEvictionStep extends AppReconcileStep {
+ /**
+ * Publishes the event if needed and always proceeds, so that the driver is
still observed.
+ *
+ * @param context The SparkAppContext for the application.
+ * @param statusRecorder The SparkAppStatusRecorder for recording status
updates.
+ * @return The ReconcileProgress to proceed with.
+ */
+ @Override
+ public ReconcileProgress reconcile(
+ SparkAppContext context, SparkAppStatusRecorder statusRecorder) {
+ Optional<Workload> workload = context.getCachedKueueWorkload();
+ // Kueue marks a deactivated Workload as evicted too, but only once it
sees the deactivation
+ Optional<Condition> eviction =
workload.flatMap(KueueWorkloadUtils::findEviction);
+ boolean deactivated =
workload.map(KueueWorkloadUtils::isDeactivated).orElse(false);
+ if (eviction.isEmpty() && !deactivated) {
+ return proceed();
+ }
+ String name = workload.get().getMetadata().getName();
+ String cause =
+ eviction.map(c -> " (" + c.getReason() + ": " + c.getMessage() +
")").orElse("");
+ EventUtils.warn(
+ context.getEventRecorder(),
+ EventUtils.REASON_KUEUE_EVICTION_IGNORED,
+ deactivated
+ ? "Kueue Workload "
+ + name
+ + " is deactivated"
+ + cause
+ + ", which the operator does not act on once the driver is
requested, so the "
+ + "driver and executors keep running although Kueue no longer
counts them against "
+ + "its quota."
+ : "Kueue evicted Workload "
+ + name
+ + 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.");
+ 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 531800d6..cb1d9db2 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
@@ -44,8 +44,8 @@ import
org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
/** Basic reconcile step for application. */
@Log4j2
public abstract sealed class AppReconcileStep
- permits AppCleanUpStep, AppInitStep, AppResourceObserveStep,
AppRunningStep,
- AppUnknownStateStep, AppValidateStep {
+ permits AppCleanUpStep, AppInitStep, AppKueueEvictionStep,
AppResourceObserveStep,
+ AppRunningStep, AppUnknownStateStep, AppValidateStep {
/**
* Reconciles a specific step for a Spark application.
*
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 1a5e3288..6dc72a8f 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
@@ -98,8 +98,8 @@ public final class EventUtils {
public static final String REASON_KUEUE_DISABLED = "KueueDisabled";
/**
- * Reason for an event describing that Kueue evicted the Workload of a
running resource in a way
- * which the operator does not act on, so the resource keeps running.
+ * Reason for an event describing that Kueue evicted or deactivated the
Workload of a running
+ * resource in a way which the operator does not act on, so the resource
keeps running.
*/
public static final String REASON_KUEUE_EVICTION_IGNORED =
"KueueEvictionIgnored";
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 10a13177..d86f7b1e 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
@@ -44,6 +44,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -85,6 +86,7 @@ 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.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.spec.ApplicationSpec;
@@ -584,4 +586,24 @@ class SparkAppReconcilerTest {
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);
+ // 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());
+ assertEquals(
+ expected ? 1 : 0,
+
steps.stream().filter(AppKueueEvictionStep.class::isInstance).count(),
+ summary.name());
+ }
+ }
}
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
new file mode 100644
index 00000000..fac5276b
--- /dev/null
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppKueueEvictionStepTest.java
@@ -0,0 +1,149 @@
+/*
+ * 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.Mockito.mock;
+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.List;
+import java.util.Optional;
+
+import io.fabric8.kubernetes.api.model.Condition;
+import io.fabric8.kubernetes.api.model.ConditionBuilder;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+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.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.ValueSource;
+import org.mockito.ArgumentCaptor;
+
+import org.apache.spark.k8s.operator.context.SparkAppContext;
+import org.apache.spark.k8s.operator.kueue.v1beta2.Workload;
+import org.apache.spark.k8s.operator.kueue.v1beta2.WorkloadSpec;
+import org.apache.spark.k8s.operator.kueue.v1beta2.WorkloadStatus;
+import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import org.apache.spark.k8s.operator.utils.EventUtils;
+import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
+
+class AppKueueEvictionStepTest {
+ private final SparkAppContext context = mock(SparkAppContext.class);
+ private final SparkAppStatusRecorder recorder =
mock(SparkAppStatusRecorder.class);
+ private final ResourceEventRecorder eventRecorder =
mock(ResourceEventRecorder.class);
+
+ @BeforeEach
+ void setUp() {
+ when(context.getEventRecorder()).thenReturn(eventRecorder);
+ }
+
+ @Test
+ void workloadWhichIsMissingOrAdmittedIsNotReported() {
+ AppKueueEvictionStep step = new AppKueueEvictionStep();
+ when(context.getCachedKueueWorkload()).thenReturn(Optional.empty());
+ Assertions.assertEquals(ReconcileProgress.proceed(),
step.reconcile(context, recorder));
+
+
when(context.getCachedKueueWorkload()).thenReturn(Optional.of(workload(true,
admitted())));
+ Assertions.assertEquals(ReconcileProgress.proceed(),
step.reconcile(context, recorder));
+
+ verifyNoInteractions(eventRecorder, recorder);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"Preempted", "PodsReadyTimeout",
"ClusterQueueStopped"})
+ void evictionIsReportedAndIgnored(String reason) {
+ when(context.getCachedKueueWorkload())
+ .thenReturn(Optional.of(workload(true, admitted(), evicted(reason))));
+
+ // The driver is still observed, e.g. until it completes
+ Assertions.assertEquals(
+ ReconcileProgress.proceed(), new
AppKueueEvictionStep().reconcile(context, recorder));
+
+ String message = captureMessages(1).get(0);
+ Assertions.assertTrue(
+ message.contains("(" + reason + ": " + reason + " by the test)"),
message);
+ // Kueue keeps counting the quota of an evicted Workload until the driver
and executors are
+ // released
+ Assertions.assertTrue(message.contains("holding its quota"), message);
+ verifyNoInteractions(recorder);
+ }
+
+ @Test
+ void deactivationIsReportedAndIgnored() {
+ when(context.getCachedKueueWorkload())
+ // Before Kueue marks the deactivated Workload as evicted
+ .thenReturn(Optional.of(workload(false, admitted())))
+ .thenReturn(Optional.of(workload(false, admitted(),
evicted("Deactivated"))))
+ // Reactivated, while Kueue keeps the Evicted condition
+ .thenReturn(Optional.of(workload(true, admitted(),
evicted("Deactivated"))));
+ AppKueueEvictionStep step = new AppKueueEvictionStep();
+ for (int i = 0; i < 3; i++) {
+ Assertions.assertEquals(ReconcileProgress.proceed(),
step.reconcile(context, recorder));
+ }
+
+ List<String> messages = captureMessages(3);
+ // Kueue stops counting the quota of a deactivated Workload
+ Assertions.assertTrue(messages.get(0).contains("is deactivated, "));
+ Assertions.assertTrue(messages.get(0).contains("no longer counts"));
+ Assertions.assertTrue(
+ messages.get(1).contains("is deactivated (Deactivated: Deactivated by
the test)"));
+ Assertions.assertTrue(messages.get(1).contains("no longer counts"));
+ // And counts it again once it is reactivated
+ Assertions.assertTrue(messages.get(2).contains("holding its quota"));
+ verifyNoInteractions(recorder);
+ }
+
+ /** Returns the messages of the published events, which are all
KueueEvictionIgnored warnings. */
+ private List<String> captureMessages(int count) {
+ ArgumentCaptor<EventRecord> captor =
ArgumentCaptor.forClass(EventRecord.class);
+ verify(eventRecorder, times(count)).record(captor.capture());
+ for (EventRecord event : captor.getAllValues()) {
+ Assertions.assertEquals(EventType.WARNING, event.type());
+ Assertions.assertEquals(EventUtils.REASON_KUEUE_EVICTION_IGNORED,
event.reason());
+ }
+ return captor.getAllValues().stream().map(EventRecord::message).toList();
+ }
+
+ private static Workload workload(boolean active, Condition... conditions) {
+ Workload workload = new Workload();
+ workload.setMetadata(new
ObjectMetaBuilder().withName("sparkapplication-app1").build());
+ workload.setSpec(WorkloadSpec.builder().active(active).build());
+
workload.setStatus(WorkloadStatus.builder().conditions(List.of(conditions)).build());
+ return workload;
+ }
+
+ private static Condition admitted() {
+ return new
ConditionBuilder().withType("Admitted").withStatus("True").build();
+ }
+
+ private static Condition evicted(String reason) {
+ return new ConditionBuilder()
+ .withType("Evicted")
+ .withStatus("True")
+ .withReason(reason)
+ .withMessage(reason + " by the test")
+ .build();
+ }
+}
diff --git a/tests/e2e/kueue/chainsaw-test.yaml
b/tests/e2e/kueue/chainsaw-test.yaml
index 76caae8b..b18e2b1d 100644
--- a/tests/e2e/kueue/chainsaw-test.yaml
+++ b/tests/e2e/kueue/chainsaw-test.yaml
@@ -954,3 +954,99 @@ spec:
kubectl delete localqueue "$LOCAL_QUEUE" -n default
--ignore-not-found=true --wait=false
kubectl delete clusterqueue "$CLUSTER_QUEUE" --ignore-not-found=true
--wait=false
kubectl delete resourceflavor "$FLAVOR" --ignore-not-found=true
--wait=false
+ - name: spark-application-whose-executors-are-not-ready-is-evicted-by-kueue
+ # Each step uses its own queue objects, so that a step never waits for
Kueue to finish
+ # deleting the objects of another step.
+ bindings:
+ - name: FLAVOR
+ value: spark-e2e-app-pods-ready-flavor
+ - name: CLUSTER_QUEUE
+ value: spark-e2e-app-pods-ready-cluster-queue
+ - name: LOCAL_QUEUE
+ value: spark-e2e-app-pods-ready-queue
+ try:
+ # The eviction below needs the waitForPodsReady timeout which the CI
shortens to 2 minutes,
+ # rather than the default of 30 minutes.
+ - script:
+ timeout: 30s
+ content: |
+ kubectl get configmap kueue-manager-config -n kueue-system -o json \
+ | jq -r '.data["controller_manager_config.yaml"] | if
contains("\nwaitForPodsReady:\n timeout: 2m\n") then "SHORTENED" else
"DEFAULT" end'
+ check:
+ (contains($stdout, 'SHORTENED')): true
+ - apply:
+ file: kueue-queues.yaml
+ # The quota fits the application (1 CPU for the driver and 1 for the
executor)
+ - script:
+ timeout: 30s
+ env:
+ - name: CLUSTER_QUEUE
+ value: ($CLUSTER_QUEUE)
+ content: |
+ kubectl patch clusterqueue "$CLUSTER_QUEUE" --type json -p \
+ '[{"op": "replace", "path":
"/spec/resourceGroups/0/flavors/0/resources/0/nominalQuota", "value": "2"},
+ {"op": "replace", "path":
"/spec/resourceGroups/0/flavors/0/resources/1/nominalQuota", "value": "8Gi"}]'
+ - apply:
+ file: spark-example-executors-not-ready.yaml
+ - assert:
+ timeout: 5m
+ resource:
+ apiVersion: kueue.x-k8s.io/v1beta2
+ kind: Workload
+ metadata:
+ name: sparkapplication-spark-job-kueue-executors-not-ready-test
+ namespace: default
+ status:
+ (conditions[?type == 'Evicted' && status == 'True' && reason ==
'PodsReadyTimeout'] | length(@)): 1
+ # A running application is not released on it yet
+ - assert:
+ timeout: 60s
+ resource:
+ apiVersion: v1
+ kind: Event
+ metadata:
+ namespace: default
+ involvedObject:
+ kind: SparkApplication
+ name: spark-job-kueue-executors-not-ready-test
+ type: Warning
+ reason: KueueEvictionIgnored
+ (contains(message || '', 'PodsReadyTimeout')): true
+ - assert:
+ timeout: 30s
+ resource:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ metadata:
+ name: spark-job-kueue-executors-not-ready-test
+ namespace: default
+ status:
+ currentState:
+ currentStateSummary: RunningHealthy
+ catch:
+ - describe:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ namespace: default
+ - describe:
+ apiVersion: kueue.x-k8s.io/v1beta2
+ kind: Workload
+ namespace: default
+ - events:
+ namespace: default
+ finally:
+ - script:
+ timeout: 120s
+ env:
+ - name: FLAVOR
+ value: ($FLAVOR)
+ - name: CLUSTER_QUEUE
+ value: ($CLUSTER_QUEUE)
+ - name: LOCAL_QUEUE
+ value: ($LOCAL_QUEUE)
+ content: |
+ kubectl delete sparkapplication
spark-job-kueue-executors-not-ready-test -n default --ignore-not-found=true
+ # Kueue finalizers may delay the deletion, which later steps do not
depend on.
+ kubectl delete localqueue "$LOCAL_QUEUE" -n default
--ignore-not-found=true --wait=false
+ kubectl delete clusterqueue "$CLUSTER_QUEUE" --ignore-not-found=true
--wait=false
+ kubectl delete resourceflavor "$FLAVOR" --ignore-not-found=true
--wait=false
diff --git a/tests/e2e/kueue/spark-example-executors-not-ready.yaml
b/tests/e2e/kueue/spark-example-executors-not-ready.yaml
new file mode 100644
index 00000000..2aa215b1
--- /dev/null
+++ b/tests/e2e/kueue/spark-example-executors-not-ready.yaml
@@ -0,0 +1,37 @@
+#
+# 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-kueue-executors-not-ready-test
+ namespace: default
+ labels:
+ kueue.x-k8s.io/queue-name: ($LOCAL_QUEUE)
+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
+ # Kueue still admits the Workload since the label is not one of the node
labels of its
+ # ResourceFlavor. Without executor thresholds, the application is
RunningHealthy meanwhile.
+
spark.kubernetes.executor.node.selector.spark.apache.org/kueue-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]
