peter-toth commented on code in PR #837:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/837#discussion_r4034542305
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java:
##########
@@ -163,6 +165,10 @@ public ReconcileProgress reconcile(
for (HasMetadata resource : resourcesToRemove) {
ReconcilerUtils.deleteResourceIfExists(context.getClient(), resource,
forceDelete);
}
+ if (KueueWorkloadFactory.hasQueueName(application)) {
Review Comment:
**Finding 1.** There is one path between the cleanup decision and this line
that returns without reaching it.
`isReleasingResourcesForSchedulingFailureAttempt` re-computes the spec, and
when that throws:
```java
} catch (Exception e) {
...
ApplicationState updatedState =
new ApplicationState(
ApplicationStateSummary.ResourceReleased,
"Cannot build Spark spec for given application, "
+ "consider all resources as released.");
...
return appendStateAndRequeueAfter( // :157, above this line
```
That is not a dead branch. It fires only when the last observed state before
termination was `SchedulingFailure` - i.e. precisely when the spec was already
rejected once - so a re-build failing the second time is the expected case, not
the exotic one.
And it is terminal. The state it writes is `ResourceReleased`, so on every
later reconcile `checkEarlyExitForTerminatedApp` takes its `:237-241` branch
and returns before the release can run. `ttlAfterStopMillis` defaults to `-1L`,
so the CR is never garbage collected either, and the `ownerReference` never
fires. The `Workload` stays `Admitted` and holds the `ClusterQueue` quota until
someone deletes the app by hand - the exact scenario the "Why" section says
this PR prevents.
I confirmed it with a probe test: a labelled app whose last state is
`SchedulingFailure` and whose `getDriverPreResourcesSpec()` throws, asserting
`KueueWorkloadUtils` is never touched. It passes on this head.
Hoisting the block above `resourcesToRemove` fixes it, and is safe because
the release does not depend on which resources get deleted:
```java
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<>();
```
I applied exactly that and re-ran `AppCleanUpStepTest`: the probe flips to
failing, which is the point, and all 17 pre-existing tests stay green -
including the three `verify(mockApp).getMetadata()` assertions you added, since
the call count is unchanged. It also closes the same hole for a
`deleteResourceIfExists` that throws, and it makes the comment you added to
those three tests literally true, since the label lookup would then really
happen before the resources are released.
Returning the quota a moment before the driver pod finishes terminating is
fine: `ClusterQueue` quota is Kueue's bookkeeping, not a kube-scheduler
reservation, and Kueue's own integrations release on job completion while pods
drain asynchronously.
##########
spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStepTest.java:
##########
@@ -115,6 +118,50 @@ 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();
Review Comment:
**Finding 2.** These two tests pin the label check in both directions, which
is the right pair. What is not pinned is the other half of the contract the
description states:
> Apps that keep their resources via `resourceRetainPolicy` keep their quota
until the resources are released.
Today that holds structurally - `:129` and `:281` both return above the
release - so there is nothing to catch. It becomes worth a test the moment
finding 1 is applied, because that change moves the release upwards and the
only thing keeping the retain guarantee is that it stays below `:113-132`. A
third sibling of these two would guard it:
```java
@Test
void cleanupWithRetainPolicyKeepsKueueWorkload() {
SparkAppStatusRecorder mockRecorder = mock(SparkAppStatusRecorder.class);
AppCleanUpStep cleanUpStep = new AppCleanUpStep(); // not on demand
SparkApplication app = new SparkApplication();
app.setMetadata(
new ObjectMetaBuilder()
.withName("app1")
.withNamespace("default")
.withLabels(Map.of(Constants.LABEL_QUEUE_NAME, "test-queue"))
.build());
app.getSpec().getApplicationTolerations().setResourceRetainPolicy(ResourceRetainPolicy.Always);
app.getSpec().getApplicationTolerations().setRestartConfig(
new RestartConfig(RestartPolicy.Never, 0, 0L, 0L));
app.setStatus(prepareApplicationStatus(ApplicationStateSummary.Failed));
...
try (MockedStatic<KueueWorkloadUtils> kueue =
Mockito.mockStatic(KueueWorkloadUtils.class)) {
cleanUpStep.reconcile(mockAppContext, mockRecorder);
kueue.verifyNoInteractions();
}
}
```
`Failed` is `isStopping()` and `Always` retains, so with
`RestartPolicy.Never` the step should take the
`TerminatedWithoutReleaseResources` branch and never touch the `Workload`.
Adjust the tolerations wiring to whatever the setters look like - the shape is
what matters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
