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 b03e313 [SPARK-57497] Fix NPE in AppInitStep restart backoff check
when stateTransitionHistory is null
b03e313 is described below
commit b03e313617f249ff314fdf95b78080db409629e7
Author: Zhou JIANG <[email protected]>
AuthorDate: Thu Jun 18 09:54:42 2026 -0700
[SPARK-57497] Fix NPE in AppInitStep restart backoff check when
stateTransitionHistory is null
### What changes were proposed in this pull request?
In AppInitStep.reconcile(), when a previous attempt summary exists, the
restart backoff
check now handles the case where
previousAttemptSummary.getStateTransitionHistory() is
null (non-trim mode). Instead of dereferencing it directly (causing an
NPE), the code
falls back to the main status.getStateTransitionHistory() and selects the
state entered
immediately before the current initializing state — the stopping state
that triggered the
restart — using NavigableMap.lowerEntry().
A null guard for getRestartConfig() was also added to the same block.
### Why are the changes needed?
In non-trim mode (TRIM_ATTEMPT_STATE_TRANSITION_HISTORY=false),
ApplicationStatus.terminateOrRestart() sets previousAttemptSummary to the
old
currentAttemptSummary, which always carries stateTransitionHistory =
null. The previous
code called .getStateTransitionHistory().get(...) unconditionally,
throwing a
NullPointerException on every restart attempt in non-trim mode. This
caused the operator
to fail to re-initialize the driver after a restart.
### Does this PR introduce any user-facing change?
No functional behavior change for users running in trim mode (the
default). In non-trim
mode, restart backoff is now correctly enforced instead of crashing with
an NPE.
### How was this patch tested?
Two unit tests were added to AppInitStepTest:
- nonTrimModeRestartBackoffElapsedProceedsToDriverCreation: verifies that
when the backoff has elapsed, the app reaches DriverRequested state without
throwing an NPE (non-trim mode, previousAttemptSummary present with null
stateTransitionHistory).
- nonTrimModeRestartBackoffActiveRequeuesWithDelay: verifies that when
the backoff has not yet elapsed, the reconcile returns a delayed requeue
without throwing an NPE, and the state remains ScheduledToRestart.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 4.6
Closes #711 from jiangzho/corner_npe.
Authored-by: Zhou JIANG <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../reconciler/reconcilesteps/AppInitStep.java | 28 ++++--
.../reconciler/reconcilesteps/AppInitStepTest.java | 102 +++++++++++++++++++++
2 files changed, 124 insertions(+), 6 deletions(-)
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
index 01c52b4..8479aad 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStep.java
@@ -27,7 +27,10 @@ import static
org.apache.spark.k8s.operator.utils.SparkExceptionUtils.buildGener
import java.time.Duration;
import java.time.Instant;
import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
import java.util.Optional;
+import java.util.SortedMap;
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.Pod;
@@ -38,6 +41,7 @@ import org.apache.spark.k8s.operator.SparkApplication;
import org.apache.spark.k8s.operator.context.SparkAppContext;
import org.apache.spark.k8s.operator.decorators.DriverResourceDecorator;
import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import org.apache.spark.k8s.operator.spec.RestartConfig;
import org.apache.spark.k8s.operator.status.ApplicationAttemptSummary;
import org.apache.spark.k8s.operator.status.ApplicationState;
import org.apache.spark.k8s.operator.status.ApplicationStateSummary;
@@ -66,14 +70,26 @@ public final class AppInitStep extends AppReconcileStep {
if (app.getStatus().getPreviousAttemptSummary() != null) {
Instant lastTransitionTime =
Instant.parse(currentState.getLastTransitionTime());
ApplicationAttemptSummary attemptSummary =
app.getStatus().getPreviousAttemptSummary();
- ApplicationState lastState = attemptSummary.getStateTransitionHistory()
- .get(attemptSummary.getStateTransitionHistory().lastKey());
+ SortedMap<Long, ApplicationState> attemptHistory =
attemptSummary.getStateTransitionHistory();
+ final ApplicationState lastState;
+ if (attemptHistory != null && !attemptHistory.isEmpty()) {
+ lastState = attemptHistory.get(attemptHistory.lastKey());
+ } else {
+ // Non-trim mode: previousAttemptSummary carries null
stateTransitionHistory.
+ // Fall back to the main history and pick the state entered just
before the current
+ // initializing state (which is the stopping state that triggered the
restart).
+ NavigableMap<Long, ApplicationState> mainNav =
+ (NavigableMap<Long, ApplicationState>)
app.getStatus().getStateTransitionHistory();
+ Map.Entry<Long, ApplicationState> prevEntry =
mainNav.lowerEntry(mainNav.lastKey());
+ lastState = prevEntry != null ? prevEntry.getValue() :
mainNav.lastEntry().getValue();
+ }
+ RestartConfig restartConfig =
+ app.getSpec().getApplicationTolerations().getRestartConfig();
+ assert restartConfig != null : "restartConfig must not be null";
Instant restartTime =
lastTransitionTime.plusMillis(
- app.getSpec()
- .getApplicationTolerations()
- .getRestartConfig()
-
.getEffectiveRestartBackoffMillis(lastState.getCurrentStateSummary()));
+ restartConfig.getEffectiveRestartBackoffMillis(
+ lastState.getCurrentStateSummary()));
Instant now = Instant.now();
if (restartTime.isAfter(now)) {
return ReconcileProgress.completeAndRequeueAfter(Duration.between(now,
restartTime));
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java
index 7f5af21..d3c7fba 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppInitStepTest.java
@@ -25,8 +25,10 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.time.Instant;
import java.util.List;
import java.util.Map;
+import java.util.TreeMap;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import io.fabric8.kubernetes.api.model.ConfigMap;
@@ -50,7 +52,13 @@ import org.mockito.ArgumentCaptor;
import org.apache.spark.k8s.operator.SparkApplication;
import org.apache.spark.k8s.operator.context.SparkAppContext;
import org.apache.spark.k8s.operator.reconciler.ReconcileProgress;
+import org.apache.spark.k8s.operator.spec.ApplicationTolerations;
import org.apache.spark.k8s.operator.spec.DeploymentMode;
+import org.apache.spark.k8s.operator.spec.RestartConfig;
+import org.apache.spark.k8s.operator.status.ApplicationAttemptSummary;
+import org.apache.spark.k8s.operator.status.ApplicationState;
+import org.apache.spark.k8s.operator.status.ApplicationStateSummary;
+import org.apache.spark.k8s.operator.status.ApplicationStatus;
import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
@EnableKubernetesMockClient(crud = true)
@@ -240,4 +248,98 @@ class AppInitStepTest {
ReconcileProgress progress =
appValidateStep.reconcile(mocksparkAppContext, recorder);
Assertions.assertEquals(ReconcileProgress.completeAndImmediateRequeue(),
progress);
}
+
+ @Test
+ void nonTrimModeRestartBackoffElapsedProceedsToDriverCreation() {
+ // Non-trim mode: previousAttemptSummary has null stateTransitionHistory.
+ // The fix falls back to the main history to resolve the stopping state
before
+ // ScheduledToRestart. With backoff elapsed the app should reach
DriverRequested without NPE.
+ AppInitStep appInitStep = new AppInitStep();
+ SparkAppContext mockContext = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+
+ SparkApplication application = new SparkApplication();
+ application.setMetadata(applicationMetadata);
+ application.getSpec().setApplicationTolerations(
+ ApplicationTolerations.builder()
+
.restartConfig(RestartConfig.builder().restartBackoffMillis(5000L).build())
+ .build());
+
+ // Main history: DriverStartTimedOut → ScheduledToRestart entered 60s ago
(backoff 5s elapsed)
+ ApplicationState timedOutState =
+ new ApplicationState(ApplicationStateSummary.DriverStartTimedOut,
"timed out");
+ ApplicationState scheduledState =
+ new ApplicationState(ApplicationStateSummary.ScheduledToRestart,
"restarting");
+
scheduledState.setLastTransitionTime(Instant.now().minusMillis(60000L).toString());
+ Map<Long, ApplicationState> history = new TreeMap<>();
+ history.put(0L, timedOutState);
+ history.put(1L, scheduledState);
+
+ // Non-trim mode: previousAttemptSummary is present but has null
stateTransitionHistory
+ ApplicationStatus status = new ApplicationStatus(
+ scheduledState, history,
+ new ApplicationAttemptSummary(), new ApplicationAttemptSummary());
+ application.setStatus(status);
+
+ when(mockContext.getResource()).thenReturn(application);
+ when(mockContext.getDriverPreResourcesSpec()).thenReturn(List.of());
+ when(mockContext.getDriverPodSpec()).thenReturn(driverPodSpec);
+ when(mockContext.getDriverResourcesSpec()).thenReturn(List.of());
+ when(mockContext.getClient()).thenReturn(kubernetesClient);
+ when(recorder.persistStatus(any(), any())).thenAnswer(invocation -> {
+ ApplicationStatus newStatus = invocation.getArgument(1);
+ application.setStatus(newStatus);
+ return true;
+ });
+
+ ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
+
+ Assertions.assertEquals(ReconcileProgress.completeAndDefaultRequeue(),
progress);
+ Assertions.assertEquals(
+ ApplicationStateSummary.DriverRequested,
+ application.getStatus().getCurrentState().getCurrentStateSummary());
+ }
+
+ @Test
+ void nonTrimModeRestartBackoffActiveRequeuesWithDelay() {
+ // Non-trim mode: backoff has NOT elapsed — should requeue with remaining
delay,
+ // not throw NullPointerException.
+ AppInitStep appInitStep = new AppInitStep();
+ SparkAppContext mockContext = mock(SparkAppContext.class);
+ SparkAppStatusRecorder recorder = mock(SparkAppStatusRecorder.class);
+
+ SparkApplication application = new SparkApplication();
+ application.setMetadata(applicationMetadata);
+ application.getSpec().setApplicationTolerations(
+ ApplicationTolerations.builder()
+
.restartConfig(RestartConfig.builder().restartBackoffMillis(60000L).build())
+ .build());
+
+ // ScheduledToRestart entered just now — 60s backoff has not elapsed
+ ApplicationState timedOutState =
+ new ApplicationState(ApplicationStateSummary.DriverStartTimedOut,
"timed out");
+ ApplicationState scheduledState =
+ new ApplicationState(ApplicationStateSummary.ScheduledToRestart,
"restarting");
+ Map<Long, ApplicationState> history = new TreeMap<>();
+ history.put(0L, timedOutState);
+ history.put(1L, scheduledState);
+
+ ApplicationStatus status = new ApplicationStatus(
+ scheduledState, history,
+ new ApplicationAttemptSummary(), new ApplicationAttemptSummary());
+ application.setStatus(status);
+
+ when(mockContext.getResource()).thenReturn(application);
+
+ ReconcileProgress progress = appInitStep.reconcile(mockContext, recorder);
+
+ // Should requeue after the remaining backoff, not throw NPE
+ Assertions.assertTrue(progress.isCompleted());
+ Assertions.assertTrue(progress.isRequeue());
+ Assertions.assertTrue(progress.getRequeueAfterDuration().toMillis() > 0);
+ // State must remain ScheduledToRestart — no driver creation attempted
+ Assertions.assertEquals(
+ ApplicationStateSummary.ScheduledToRestart,
+ application.getStatus().getCurrentState().getCurrentStateSummary());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
