(spark-kubernetes-operator) branch main updated: [SPARK-59594] Release Kueue `Workload` when `AppCleanUpStep` releases `SparkApplication` resources

[do...@apache.org]

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 6920cc52 [SPARK-59594] Release Kueue `Workload` when `AppCleanUpStep` releases `SparkApplication` resources
6920cc52 is described below

commit 6920cc528ecd30dfbe60f8b76b4757ed5bfba611
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Sep 17 03:15:37 2026 -0700

    [SPARK-59594] Release Kueue `Workload` when `AppCleanUpStep` releases `SparkApplication` resources
    
    ### What changes were proposed in this pull request?
    
    This PR aims to delete the Kueue `Workload` when `AppCleanUpStep` releases the resources of a `SparkApplication` with the `kueue.x-k8s.io/queue-name` label.
    
    The `Workload` is deleted before the attempt's resources, so it is released on every path that reaches the resource release:
    - On-demand clean-up
    - Stopping apps, including before a restart
    - Terminated apps whose retention period has expired
    - `SchedulingFailure` attempts whose spec cannot be built again
    
    Apps that keep their resources via `resourceRetainPolicy` keep their quota until the resources are released.
    
    ### Why are the changes needed?
    
    This is part of the Kueue runtime integration. Once labeled apps wait for admission, a `Workload` left `Admitted` after the app finishes holds the `ClusterQueue` quota indefinitely by default (`ttlAfterStopMillis=-1`). This blocks other jobs in the same `LocalQueue`.
    
    - Deleting the `Workload` returns the quota and needs no `workloads/status` permission. A restarted attempt is queued again with a new `Workload`, so the quota is not held during the restart backoff.
    - Failures are only logged as warnings, and a missing `Workload` or Kueue CRD is a no-op. So clean-up is never blocked, and `ownerReference` GC still removes the `Workload` eventually.
    - This should land before the admission PR so that `main` never leaks quota.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No, the operator does not create Kueue Workloads yet.
    
    ### How was this patch tested?
    
    Pass the CIs with the newly added test cases and the updated existing test cases in `AppCleanUpStepTest`. The new test cases cover apps with and without the queue label, the `SchedulingFailure` spec-build failure path, and the `resourceRetainPolicy` path.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Claude Opus 5
    
    Closes #837 from dongjoon-hyun/SPARK-59594.
    
    Authored-by: Dongjoon Hyun <do...@apache.org>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../reconciler/reconcilesteps/AppCleanUpStep.java  |   6 ++
 .../reconcilesteps/AppCleanUpStepTest.java         | 112 +++++++++++++++++++++
 2 files changed, 118 insertions(+)

diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java
index e0dc065a..5d536ac0 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java
@@ -37,6 +37,8 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.spark.k8s.operator.SparkApplication;
 import org.apache.spark.k8s.operator.config.SparkOperatorConf;
 import org.apache.spark.k8s.operator.context.SparkAppContext;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadFactory;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
 import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
 import org.apache.spark.k8s.operator.spec.ApplicationTolerations;
 import org.apache.spark.k8s.operator.spec.ResourceRetainPolicy;
@@ -133,6 +135,10 @@ public final class AppCleanUpStep extends AppReconcileStep {
       return ReconcileProgress.proceed();
     }
 
+    if (KueueWorkloadFactory.hasQueueName(application)) {
+      // Release the quota. A restarted attempt is queued again with a new Workload.
+      KueueWorkloadUtils.releaseWorkload(context.getClient(), application);
+    }
     List<HasMetadata> resourcesToRemove = new ArrayList<>();
     if (isReleasingResourcesForSchedulingFailureAttempt(currentStatus)) {
       // if app failed at scheduling, re-compute all spec and delete as they may not be fully
diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStepTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStepTest.java
index 86f30867..a911428f 100644
--- a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStepTest.java
+++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStepTest.java
@@ -32,9 +32,11 @@ import static org.mockito.Mockito.when;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.fabric8.kubernetes.api.model.Pod;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import org.junit.jupiter.api.Assertions;
@@ -46,6 +48,7 @@ import org.mockito.Mockito;
 import org.apache.spark.k8s.operator.Constants;
 import org.apache.spark.k8s.operator.SparkApplication;
 import org.apache.spark.k8s.operator.context.SparkAppContext;
+import org.apache.spark.k8s.operator.kueue.KueueWorkloadUtils;
 import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
 import org.apache.spark.k8s.operator.spec.ApplicationSpec;
 import org.apache.spark.k8s.operator.spec.ApplicationTolerations;
@@ -115,6 +118,108 @@ class AppCleanUpStepTest {
           notExceedRetainDuration,
           notExceedTtl);
 
+  @Test
+  void cleanupReleasesKueueWorkload() {
+    SparkAppStatusRecorder mockRecorder = mock(SparkAppStatusRecorder.class);
+    AppCleanUpStep cleanUpWithReason = new AppCleanUpStep(SparkAppStatusUtils::appCancelled);
+    SparkApplication app = new SparkApplication();
+    app.setMetadata(
+        new ObjectMetaBuilder()
+            .withName("app1")
+            .withNamespace("default")
+            .withLabels(Map.of(Constants.LABEL_QUEUE_NAME, "test-queue"))
+            .build());
+    app.setStatus(prepareApplicationStatus(ApplicationStateSummary.RunningHealthy));
+    SparkAppContext mockAppContext = mock(SparkAppContext.class);
+    when(mockAppContext.getResource()).thenReturn(app);
+    KubernetesClient mockClient = mock(KubernetesClient.class);
+    when(mockAppContext.getClient()).thenReturn(mockClient);
+    when(mockAppContext.getDriverPod()).thenReturn(Optional.empty());
+    when(mockRecorder.appendNewStateAndPersist(eq(mockAppContext), any())).thenReturn(true);
+
+    try (MockedStatic<KueueWorkloadUtils> kueue = Mockito.mockStatic(KueueWorkloadUtils.class)) {
+      cleanUpWithReason.reconcile(mockAppContext, mockRecorder);
+      kueue.verify(() -> KueueWorkloadUtils.releaseWorkload(mockClient, app));
+    }
+  }
+
+  @Test
+  void cleanupWithoutQueueNameDoesNotReleaseKueueWorkload() {
+    SparkAppStatusRecorder mockRecorder = mock(SparkAppStatusRecorder.class);
+    AppCleanUpStep cleanUpWithReason = new AppCleanUpStep(SparkAppStatusUtils::appCancelled);
+    SparkApplication app = new SparkApplication();
+    app.setMetadata(new ObjectMetaBuilder().withName("app1").withNamespace("default").build());
+    app.setStatus(prepareApplicationStatus(ApplicationStateSummary.RunningHealthy));
+    SparkAppContext mockAppContext = mock(SparkAppContext.class);
+    when(mockAppContext.getResource()).thenReturn(app);
+    when(mockAppContext.getClient()).thenReturn(mock(KubernetesClient.class));
+    when(mockAppContext.getDriverPod()).thenReturn(Optional.empty());
+    when(mockRecorder.appendNewStateAndPersist(eq(mockAppContext), any())).thenReturn(true);
+
+    try (MockedStatic<KueueWorkloadUtils> kueue = Mockito.mockStatic(KueueWorkloadUtils.class)) {
+      cleanUpWithReason.reconcile(mockAppContext, mockRecorder);
+      kueue.verifyNoInteractions();
+    }
+  }
+
+  @Test
+  void cleanupReleasesKueueWorkloadWhenSpecBuildFails() {
+    SparkAppStatusRecorder mockRecorder = mock(SparkAppStatusRecorder.class);
+    AppCleanUpStep routineCheck = new AppCleanUpStep();
+    SparkApplication app = new SparkApplication();
+    app.setMetadata(
+        new ObjectMetaBuilder()
+            .withName("app1")
+            .withNamespace("default")
+            .withLabels(Map.of(Constants.LABEL_QUEUE_NAME, "test-queue"))
+            .build());
+    app.setStatus(prepareApplicationStatus(ApplicationStateSummary.SchedulingFailure));
+    SparkAppContext mockAppContext = mock(SparkAppContext.class);
+    when(mockAppContext.getResource()).thenReturn(app);
+    KubernetesClient mockClient = mock(KubernetesClient.class);
+    when(mockAppContext.getClient()).thenReturn(mockClient);
+    when(mockAppContext.getDriverPreResourcesSpec()).thenThrow(new IllegalStateException("foo"));
+    when(mockRecorder.appendNewStateAndPersist(eq(mockAppContext), any())).thenReturn(true);
+
+    try (MockedStatic<KueueWorkloadUtils> kueue = Mockito.mockStatic(KueueWorkloadUtils.class)) {
+      routineCheck.reconcile(mockAppContext, mockRecorder);
+      kueue.verify(() -> KueueWorkloadUtils.releaseWorkload(mockClient, app));
+    }
+    ArgumentCaptor<ApplicationState> captor = ArgumentCaptor.forClass(ApplicationState.class);
+    verify(mockRecorder).appendNewStateAndPersist(eq(mockAppContext), captor.capture());
+    Assertions.assertEquals(
+        ApplicationStateSummary.ResourceReleased, captor.getValue().getCurrentStateSummary());
+  }
+
+  @Test
+  void cleanupWithRetainPolicyKeepsKueueWorkload() {
+    SparkAppStatusRecorder mockRecorder = mock(SparkAppStatusRecorder.class);
+    AppCleanUpStep routineCheck = new AppCleanUpStep();
+    SparkApplication app = new SparkApplication();
+    app.setMetadata(
+        new ObjectMetaBuilder()
+            .withName("app1")
+            .withNamespace("default")
+            .withLabels(Map.of(Constants.LABEL_QUEUE_NAME, "test-queue"))
+            .build());
+    app.setSpec(alwaysRetain);
+    app.setStatus(prepareApplicationStatus(ApplicationStateSummary.Failed));
+    SparkAppContext mockAppContext = mock(SparkAppContext.class);
+    when(mockAppContext.getResource()).thenReturn(app);
+    when(mockAppContext.getClient()).thenReturn(mock(KubernetesClient.class));
+    when(mockRecorder.appendNewStateAndPersist(eq(mockAppContext), any())).thenReturn(true);
+
+    try (MockedStatic<KueueWorkloadUtils> kueue = Mockito.mockStatic(KueueWorkloadUtils.class)) {
+      routineCheck.reconcile(mockAppContext, mockRecorder);
+      kueue.verifyNoInteractions();
+    }
+    ArgumentCaptor<ApplicationState> captor = ArgumentCaptor.forClass(ApplicationState.class);
+    verify(mockRecorder).appendNewStateAndPersist(eq(mockAppContext), captor.capture());
+    Assertions.assertEquals(
+        ApplicationStateSummary.TerminatedWithoutReleaseResources,
+        captor.getValue().getCurrentStateSummary());
+  }
+
   @Test
   void enableForceDelete() {
     AppCleanUpStep appCleanUpStep = new AppCleanUpStep();
@@ -192,6 +297,8 @@ class AppCleanUpStepTest {
         Assertions.assertEquals(
             ApplicationStateSummary.ResourceReleased, appState.getCurrentStateSummary());
         Assertions.assertEquals(Constants.APP_CANCELLED_MESSAGE, appState.getMessage());
+        // The Kueue queue label is looked up before releasing the resources
+        verify(mockApp).getMetadata();
         verifyNoMoreInteractions(mockAppContext, mockRecorder, mockApp, mockClient, driverPod);
       }
     }
@@ -282,6 +389,8 @@ class AppCleanUpStepTest {
     Assertions.assertEquals(
         ApplicationStateSummary.ResourceReleased, appState.getCurrentStateSummary());
     Assertions.assertEquals(Constants.APP_CANCELLED_MESSAGE, appState.getMessage());
+    // The Kueue queue label is looked up before releasing the resources
+    verify(mockApp).getMetadata();
     verifyNoMoreInteractions(mockAppContext, mockRecorder, mockApp, mockClient, driverPod);
   }
 
@@ -360,6 +469,9 @@ class AppCleanUpStepTest {
     Assertions.assertEquals(
         ApplicationStateSummary.ResourceReleased, appState2.getCurrentStateSummary());
     Assertions.assertEquals(Constants.APP_CANCELLED_MESSAGE, appState2.getMessage());
+    // The Kueue queue label is looked up before releasing the resources
+    verify(mockApp1).getMetadata();
+    verify(mockApp2).getMetadata();
     verifyNoMoreInteractions(
         mockAppContext1, mockAppContext2, mockRecorder, mockApp1, mockApp2, mockClient, driverPod);
   }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org