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 eb66854a [SPARK-60018] Add `ApplicationStateSummary.Suspended` and
`ApplicationStatus#resume`
eb66854a is described below
commit eb66854a61f79b99b0244d6d3a5f19b056892d1e
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Oct 6 23:13:59 2026 -0700
[SPARK-60018] Add `ApplicationStateSummary.Suspended` and
`ApplicationStatus#resume`
### What changes were proposed in this pull request?
This PR aims to add `ApplicationStateSummary.Suspended` and
`ApplicationStatus#resume`.
- Add `Suspended` to `ApplicationStateSummary`, and make `isStopping()`
exclude it. Otherwise, `AppCleanUpStep` would end a suspended application. The
result for the existing states is unchanged.
- Add `ApplicationStatus#resume`, which starts the next attempt of a
`Suspended` application from `Submitted`. Unlike a restart, it does not
increase the restart counters, and it resets only
`schedulingFailureRestartCounter`.
- Add `Suspended` to the `SparkApplication` CRD of `spark.apache.org/v1`,
and update `docs/migration_guide.md`.
### Why are the changes needed?
To support `spec.suspend` for a running `SparkApplication` in a follow-up
PR. Currently, it takes effect only before the driver is requested.
### Does this PR introduce _any_ user-facing change?
Yes, in the API only. Unlike
[1.0.0](https://github.com/apache/spark-kubernetes-operator/releases/tag/1.0.0)
(2026-07-23), the `SparkApplication` CRD accepts the `Suspended` state, an
exhaustive `switch` over `ApplicationStateSummary` needs a case for
`Suspended`, and `ApplicationStatus` has the new public method `resume`. There
is no behavior change, since nothing uses them until the follow-up PR.
### 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 #946 from dongjoon-hyun/SPARK-60018.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../sparkapplications.spark.apache.org-v1.yaml | 4 +
docs/migration_guide.md | 8 +-
.../operator/status/ApplicationStateSummary.java | 17 ++-
.../k8s/operator/status/ApplicationStatus.java | 57 +++++++++-
.../status/ApplicationStateSummaryTest.java | 13 +++
.../k8s/operator/status/ApplicationStatusTest.java | 121 +++++++++++++++++++++
6 files changed, 209 insertions(+), 11 deletions(-)
diff --git
a/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
b/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
index f8b5aaf1..3c799123 100644
---
a/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
+++
b/build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml
@@ -16556,6 +16556,7 @@ spec:
- SchedulingFailure
- Submitted
- Succeeded
+ - Suspended
- TerminatedWithoutReleaseResources
type: string
lastObservedDriverStatus:
@@ -17231,6 +17232,7 @@ spec:
- SchedulingFailure
- Submitted
- Succeeded
+ - Suspended
- TerminatedWithoutReleaseResources
type: string
lastObservedDriverStatus:
@@ -17918,6 +17920,7 @@ spec:
- SchedulingFailure
- Submitted
- Succeeded
+ - Suspended
- TerminatedWithoutReleaseResources
type: string
lastObservedDriverStatus:
@@ -18594,6 +18597,7 @@ spec:
- SchedulingFailure
- Submitted
- Succeeded
+ - Suspended
- TerminatedWithoutReleaseResources
type: string
lastObservedDriverStatus:
diff --git a/docs/migration_guide.md b/docs/migration_guide.md
index 8a2b0c76..696804b7 100644
--- a/docs/migration_guide.md
+++ b/docs/migration_guide.md
@@ -33,8 +33,8 @@ affect building the project, CI, tests or examples.
- Since 1.1.0, `spark.apache.org/v1beta1` of the CRDs is deprecated and
scheduled for removal in
2.0.0. Its schema is no longer updated, e.g. it lacks `spec.suspend`. Use
`spark.apache.org/v1`.
- Since 1.1.0, K8s 1.35 or newer is recommended instead of 1.34.
-- Since 1.1.0, the CRDs have `spec.suspend` for `SparkApplication` and
`SparkCluster`, and the
- `Suspended` state for `SparkCluster`. Since `helm upgrade` does not update
CRDs, replace them
+- Since 1.1.0, the CRDs have `spec.suspend` and the `Suspended` state for
`SparkApplication` and
+ `SparkCluster`. Since `helm upgrade` does not update CRDs, replace them
before upgrading the operator, e.g. with the CRDs of the Helm chart `1.9.0`,
which deploys
1.1.0. Otherwise, the API server rejects or drops `spec.suspend`, and
rejects the `Suspended`
state.
@@ -209,5 +209,5 @@ is recreated, e.g. by suspending and resuming it with
[`spec.suspend`](spark_cus
is deprecated for removal. It ignores `resourceRetainPolicy`, so it no
longer returns
`TerminatedWithoutReleaseResources`. Use `terminateOrRestart(RestartConfig,
String, boolean)`
instead.
-- Since 1.1.0, `ClusterStateSummary` has the `Suspended` constant, so an
exhaustive `switch` over
- it needs a case for it.
+- Since 1.1.0, `ApplicationStateSummary` and `ClusterStateSummary` have the
`Suspended` constant,
+ so an exhaustive `switch` over either needs a case for it.
diff --git
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
index 1ef8c688..48186b6d 100644
---
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
+++
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStateSummary.java
@@ -109,7 +109,15 @@ public enum ApplicationStateSummary implements
BaseStateSummary {
* If configured, operator may mark app as terminated without releasing
resources. While this can
* be helpful in dev phase, it shall not be enabled for prod use cases
*/
- TerminatedWithoutReleaseResources;
+ TerminatedWithoutReleaseResources,
+
+ /**
+ * The application is suspended by spec.suspend after its driver was
requested, so its driver
+ * and executors are released until it is resumed with a new attempt from
Submitted. It is
+ * declared last, after the terminated states, so that the ordinal ranges of
{@link
+ * #isStarting()} and {@link #isStopping()} exclude it.
+ */
+ Suspended;
public static final Set<ApplicationStateSummary> infrastructureFailures =
Set.of(DriverStartTimedOut, ExecutorsStartTimedOut, SchedulingFailure);
@@ -161,12 +169,13 @@ public enum ApplicationStateSummary implements
BaseStateSummary {
* Checks if the application is in the `stopping` state. The operator
releases its resources based
* on retain policy and perform retry based on retry policy.
*
- * @return True if the state is past RunningWithBelowThresholdExecutors but
not yet terminated,
- * false otherwise.
+ * @return True if the state is past RunningWithBelowThresholdExecutors and
before the terminated
+ * states, false otherwise.
* @since 0.1.0
*/
public boolean isStopping() {
- return RunningWithBelowThresholdExecutors.ordinal() < this.ordinal() &&
!isTerminated();
+ return RunningWithBelowThresholdExecutors.ordinal() < this.ordinal()
+ && this.ordinal() < ResourceReleased.ordinal();
}
/**
diff --git
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStatus.java
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStatus.java
index 50e185dd..3114f4ec 100644
---
a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStatus.java
+++
b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStatus.java
@@ -205,10 +205,61 @@ public class ApplicationStatus
currentAttemptSummary);
}
- ApplicationAttemptSummary nextAttemptSummary = new
ApplicationAttemptSummary(nextAttemptInfo);
- ApplicationState state =
- new ApplicationState(ApplicationStateSummary.ScheduledToRestart,
stateMessageOverride);
+ return startNextAttempt(
+ new ApplicationState(ApplicationStateSummary.ScheduledToRestart,
stateMessageOverride),
+ nextAttemptInfo,
+ trimStateTransitionHistory);
+ }
+ /**
+ * Starts a new attempt of a suspended application which is resumed. Like a
restart, the new
+ * attempt gets the next attempt id, and the history of the suspended
attempt moves to the
+ * previous attempt summary if trimStateTransitionHistory is set. Unlike a
restart, it keeps the
+ * restart counters, since the suspended attempt neither failed nor
finished, so it does not count
+ * against the restart limits, whatever the restart policy is. Only the
counter of consecutive
+ * scheduling failures starts over, since the driver of the suspended
attempt was requested, like
+ * after an attempt which failed otherwise.
+ *
+ * <p>It only creates the status of the new attempt. Like {@link
+ * #terminateOrRestart(RestartConfig, String, boolean)}, the resources of
the suspended attempt
+ * are expected to be released already.
+ *
+ * @param stateMessage The message of the Submitted state which the new
attempt starts with.
+ * @param trimStateTransitionHistory If true, the state transition history
will be trimmed.
+ * @return An updated ApplicationStatus object.
+ * @throws IllegalStateException if the application is not Suspended.
+ * @since 1.1.0
+ */
+ public ApplicationStatus resume(String stateMessage, boolean
trimStateTransitionHistory) {
+ if (ApplicationStateSummary.Suspended !=
currentState.getCurrentStateSummary()) {
+ throw new IllegalStateException(
+ "Spark application cannot be resumed unless it is suspended, current
state is: "
+ + currentState);
+ }
+ ApplicationAttemptInfo attemptInfo =
currentAttemptSummary.getAttemptInfo();
+ return startNextAttempt(
+ new ApplicationState(ApplicationStateSummary.Submitted, stateMessage),
+ new ApplicationAttemptInfo(
+ attemptInfo.getId() + 1L,
+ attemptInfo.getRestartCounter(),
+ attemptInfo.getFailureRestartCounter(),
+ 0L),
+ trimStateTransitionHistory);
+ }
+
+ /**
+ * Starts the next attempt with the given state as its first state.
+ *
+ * @param state The first state of the next attempt.
+ * @param nextAttemptInfo The attempt info of the next attempt.
+ * @param trimStateTransitionHistory If true, the state transition history
will be trimmed.
+ * @return An updated ApplicationStatus object.
+ */
+ private ApplicationStatus startNextAttempt(
+ ApplicationState state,
+ ApplicationAttemptInfo nextAttemptInfo,
+ boolean trimStateTransitionHistory) {
+ ApplicationAttemptSummary nextAttemptSummary = new
ApplicationAttemptSummary(nextAttemptInfo);
if (trimStateTransitionHistory) {
// when truncating, put all previous history entries into previous
attempt summary
ApplicationAttemptSummary newPrevSummary =
diff --git
a/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStateSummaryTest.java
b/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStateSummaryTest.java
index 0a1e0f76..d904936f 100644
---
a/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStateSummaryTest.java
+++
b/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStateSummaryTest.java
@@ -81,6 +81,19 @@ class ApplicationStateSummaryTest {
assertFalse(ApplicationStateSummary.DriverRequested.isStopping());
}
+ @Test
+ void testSuspendedIsNeitherStoppingNorTerminated() {
+ // A suspended application is reconciled again to be resumed, so it must
not be terminated, and
+ // it must not be mistaken for a stopping attempt, which the clean up
would end, by the ordinal
+ // comparison of isStopping
+ assertFalse(ApplicationStateSummary.Suspended.isInitializing());
+ assertFalse(ApplicationStateSummary.Suspended.isStarting());
+ assertFalse(ApplicationStateSummary.Suspended.isStopping());
+ assertFalse(ApplicationStateSummary.Suspended.isTerminated());
+ assertFalse(ApplicationStateSummary.Suspended.isFailure());
+ assertFalse(ApplicationStateSummary.Suspended.isInfrastructureFailure());
+ }
+
@Test
void testIsTerminated() {
assertTrue(ApplicationStateSummary.ResourceReleased.isTerminated());
diff --git
a/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStatusTest.java
b/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStatusTest.java
index 0727163e..dcc20385 100644
---
a/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStatusTest.java
+++
b/spark-operator-api/src/test/java/org/apache/spark/k8s/operator/status/ApplicationStatusTest.java
@@ -24,6 +24,7 @@ import static
org.apache.spark.k8s.operator.status.ApplicationStateSummary.Succe
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.time.Duration;
@@ -417,6 +418,126 @@ class ApplicationStatusTest {
}
}
+ @Test
+ void testResumeStartsNextAttemptWithoutCountingARestart() {
+ // The second attempt, after a restart on a scheduling failure, runs until
it is suspended
+ ApplicationAttemptInfo attemptInfo = new ApplicationAttemptInfo(1L, 1L,
1L, 1L);
+ ApplicationStatus suspended =
+ new ApplicationStatus(
+ new
ApplicationState(ApplicationStateSummary.ScheduledToRestart, ""),
+ Map.of(4L, new
ApplicationState(ApplicationStateSummary.ScheduledToRestart, "")),
+ new ApplicationAttemptSummary(),
+ new ApplicationAttemptSummary(attemptInfo))
+ .appendNewState(new
ApplicationState(ApplicationStateSummary.DriverRequested, ""))
+ .appendNewState(new
ApplicationState(ApplicationStateSummary.RunningHealthy, ""))
+ .appendNewState(new
ApplicationState(ApplicationStateSummary.Suspended, ""));
+
+ for (boolean trimStateTransitionHistory : new boolean[] {false, true}) {
+ ApplicationStatus resumed = suspended.resume("resumed",
trimStateTransitionHistory);
+
+ ApplicationState state = resumed.getCurrentState();
+ assertEquals(Submitted, state.getCurrentStateSummary());
+ assertEquals("resumed", state.getMessage());
+ assertEquals(state, resumed.getStateTransitionHistory().get(8L));
+ // The restart counters stay, since the suspended attempt neither failed
nor finished, while
+ // its driver was requested, which ends a run of consecutive scheduling
failures
+ assertEquals(
+ new ApplicationAttemptInfo(2L, 1L, 1L, 0L),
+ resumed.getCurrentAttemptSummary().getAttemptInfo());
+ if (trimStateTransitionHistory) {
+ // Like a restart, the history of the suspended attempt moves to the
previous attempt
+ assertEquals(1, resumed.getStateTransitionHistory().size());
+ assertEquals(attemptInfo,
resumed.getPreviousAttemptSummary().getAttemptInfo());
+ assertEquals(
+ suspended.getStateTransitionHistory(),
+ resumed.getPreviousAttemptSummary().getStateTransitionHistory());
+ } else {
+ assertEquals(5, resumed.getStateTransitionHistory().size());
+ assertEquals(suspended.getCurrentAttemptSummary(),
resumed.getPreviousAttemptSummary());
+ }
+ }
+ }
+
+ @Test
+ void testOnlySuspendedApplicationIsResumed() {
+ for (ApplicationStateSummary summary : ApplicationStateSummary.values()) {
+ if (summary != ApplicationStateSummary.Suspended) {
+ ApplicationStatus status =
+ new ApplicationStatus().appendNewState(new
ApplicationState(summary, ""));
+ assertThrows(
+ IllegalStateException.class,
+ () -> assertNotNull(status.resume("", true)),
+ summary.name());
+ }
+ }
+ }
+
+ @Test
+ void testResumeEndsConsecutiveSchedulingFailures() {
+ RestartConfig config =
+ RestartConfig.builder()
+ .restartPolicy(RestartPolicy.Always)
+ .maxRestartAttempts(10L)
+ .maxRestartOnSchedulingFailure(1L)
+ .build();
+ // The first attempt fails to be scheduled, and the second one gets its
driver and runs until
+ // it is suspended
+ ApplicationStatus suspended =
+ new ApplicationStatus()
+ .appendNewState(new
ApplicationState(ApplicationStateSummary.SchedulingFailure, ""))
+ .terminateOrRestart(config, null, true)
+ .appendNewState(new
ApplicationState(ApplicationStateSummary.DriverRequested, ""))
+ .appendNewState(new
ApplicationState(ApplicationStateSummary.RunningHealthy, ""))
+ .appendNewState(new
ApplicationState(ApplicationStateSummary.Suspended, ""));
+
+ // A scheduling failure of the resumed attempt does not follow the first
one, so it restarts
+ ApplicationStatus restarted =
+ suspended
+ .resume("", true)
+ .appendNewState(new
ApplicationState(ApplicationStateSummary.SchedulingFailure, ""))
+ .terminateOrRestart(config, null, true);
+ assertEquals(
+ ApplicationStateSummary.ScheduledToRestart,
+ restarted.getCurrentState().getCurrentStateSummary());
+ }
+
+ @Test
+ void testRunBeforeSuspensionDoesNotResetRestartCounter() {
+ RestartConfig config = new RestartConfig();
+ config.setRestartPolicy(RestartPolicy.Always);
+ config.setMaxRestartAttempts(1L);
+ config.setRestartCounterResetMillis(3600000L); // 1 hour
+
+ Instant now = Instant.now();
+ Instant fourHoursAgo = now.minus(Duration.ofHours(4));
+ // After its only permitted restart, the attempt runs for two hours until
it is suspended, and
+ // stays suspended for two hours
+ ApplicationStatus suspended =
+ new ApplicationStatus(
+ stateAt(ApplicationStateSummary.ScheduledToRestart,
fourHoursAgo),
+ Map.of(4L, stateAt(ApplicationStateSummary.ScheduledToRestart,
fourHoursAgo)),
+ new ApplicationAttemptSummary(),
+ new ApplicationAttemptSummary(new ApplicationAttemptInfo(1L,
1L, 0L, 0L)))
+ .appendNewState(stateAt(ApplicationStateSummary.DriverRequested,
fourHoursAgo))
+ .appendNewState(
+ stateAt(ApplicationStateSummary.Suspended,
now.minus(Duration.ofHours(2))));
+
+ for (boolean trimStateTransitionHistory : new boolean[] {false, true}) {
+ // The resumed attempt fails right away. Neither the run before the
suspension nor the
+ // suspension counts as a run of the resumed attempt, so no more restart
is permitted.
+ ApplicationStatus failed =
+ suspended
+ .resume("", trimStateTransitionHistory)
+ .appendNewState(stateAt(ApplicationStateSummary.DriverRequested,
now))
+ .appendNewState(stateAt(ApplicationStateSummary.Failed, now));
+ ApplicationStatus terminated =
+ failed.terminateOrRestart(config, null, trimStateTransitionHistory);
+ assertEquals(
+ ApplicationStateSummary.ResourceReleased,
+ terminated.getCurrentState().getCurrentStateSummary());
+ }
+ }
+
private ApplicationState stateAt(ApplicationStateSummary summary, Instant
time) {
ApplicationState state = new ApplicationState(summary, "");
state.setLastTransitionTime(time.toString());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
