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 6469697 [SPARK-57868] Requeue after driver not found because of the
stale cache
6469697 is described below
commit 64696971d813201583f8c8e26dc8bf6ff1fd3bcc
Author: Qi Tan <[email protected]>
AuthorDate: Mon Jul 13 09:02:22 2026 -0700
[SPARK-57868] Requeue after driver not found because of the stale cache
### What changes were proposed in this pull request?
When the operator observes a SparkApplication and the driver pod is absent
from the informer (secondary-resource) cache, it previously concluded
immediately that the driver had been removed and transitioned the application
to Failed (driverUnexpectedRemoved, message "Driver removed…").
This PR makes that determination robust against transient informer-cache
staleness:
- Grace period before verification. If the driver pod is missing from the
cache but the application's current state is younger than
spark.kubernetes.operator.reconciler.missingDriverGracePeriodSeconds (default
60), the operator defers the verdict and requeues, giving the informer time to
catch up (e.g. right after pod creation or a watch reconnect).
- Authoritative live lookup. Once past the grace period, the operator
performs a direct, cache-bypassing lookup against the API server via the new
SparkAppContext.getDriverPodFromApi(). Only if the API server also reports no
driver pod is the application marked driverUnexpectedRemoved. If the pod is
present, the informer is treated as stale and the live pod is used for the
current reconcile.
- Short-lived requeue cadence. Deferrals requeue after
spark.kubernetes.operator.reconciler.defaultRequeueIntervalSeconds
(default 15), distinct from the coarse steady-state
reconciler.intervalSeconds (default 120), so the transient condition is
re-checked promptly rather than waiting for the next periodic reconcile.
- API-failure handling. If the live lookup itself throws (API server
unreachable), the removal verdict is deferred and the reconcile requeues rather
than failing the application; the condition self-heals when the API server
recovers.
Two new dynamically-overridable config options are added in
SparkOperatorConf, documented in docs/config_properties.md.
### Why are the changes needed?
An informer cache can briefly diverge from API-server state. The driver pod
appears "missing" from the cache even though it is healthy on the API server.
The previous logic treated a cache miss as ground truth and failed the
application.
### Does this PR introduce _any_ user-facing change?
yes
### How was this patch tested?
unit test
### Was this patch authored or co-authored using generative AI tooling?
Yes. Generated-by: Claude Code Opus 4.8 (1M context)
Closes #749 from TQJADE/requeue.
Authored-by: Qi Tan <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
docs/config_properties.md | 2 +
.../k8s/operator/config/SparkOperatorConf.java | 42 +++++
.../k8s/operator/context/SparkAppContext.java | 27 ++++
.../reconcilesteps/AppReconcileStep.java | 81 +++++++---
.../reconcilesteps/AppDriverPodCacheStaleTest.java | 169 +++++++++++++++++++++
5 files changed, 302 insertions(+), 19 deletions(-)
diff --git a/docs/config_properties.md b/docs/config_properties.md
index 1a9ff64..3888cb0 100644
--- a/docs/config_properties.md
+++ b/docs/config_properties.md
@@ -41,6 +41,8 @@
| spark.kubernetes.operator.reconciler.foregroundRequestTimeoutSeconds | Long
| 60 | true | Timeout (in seconds) for requests made to API server. This
applies only to foreground requests. |
| spark.kubernetes.operator.reconciler.intervalSeconds | Long | 120 | true |
Interval (in seconds, non-negative) to reconcile Spark applications. Note that
reconciliation is always expected to be triggered when app spec / status is
updated. This interval controls the reconcile behavior of operator
reconciliation even when there's no update on SparkApplication, e.g. to
determine whether a hanging app needs to be proactively terminated. Thus this
is recommended to set to above 2 minutes t [...]
| spark.kubernetes.operator.reconciler.labelSelector | String | | false | A
label selector that identifies a set of Spark custom resources for the operator
to reconcile. If not set or set to empty string, operator would reconcile all
Spark Apps and Clusters. |
+ | spark.kubernetes.operator.reconciler.missingDriverGracePeriodSeconds |
Integer | 60 | true | Grace period (in seconds) the operator waits for the pod
informer cache to catch up after the driver pod is observed missing, before
doing a live API verification against the Kubernetes apiserver. This protects
healthy applications from being failed when the informer cache is briefly stale
(e.g., after a watch reconnect). Set to 0 to verify on every cache miss. |
+ | spark.kubernetes.operator.reconciler.missingDriverRequeueIntervalSeconds |
Integer | 15 | true | Requeue interval (in seconds) used to re-check a driver
pod that is missing from the informer cache, while waiting for the cache to
catch up during the missing-driver grace period. Specific to the missing-driver
flow; this is not the operator's default requeue cadence. The steady-state
periodic reconcile interval is
spark.kubernetes.operator.reconciler.intervalSeconds. Set to 0 to requeue [...]
| spark.kubernetes.operator.reconciler.parallelism | Integer | 50 | false |
Thread pool size for Spark Operator reconcilers. Unbounded pool would be used
if set to non-positive number. |
| spark.kubernetes.operator.reconciler.rateLimiter.maxLoopForPeriod | Integer
| 5 | false | Max number of reconcile loops triggered within the rate limiter
refresh period for each resource. Setting the limit <= 0 disables the limiter.
|
| spark.kubernetes.operator.reconciler.rateLimiter.refreshPeriodSeconds |
Integer | 15 | false | Operator rate limiter refresh period(in seconds) for
each resource. |
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
index a1a4c10..d6d3202 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
@@ -220,6 +220,48 @@ public final class SparkOperatorConf {
.defaultValue(120L)
.build();
+ /**
+ * Grace period (in seconds) the operator waits for the pod informer cache
to catch up
+ * after the driver pod is observed missing, before doing a live API
verification. Brief
+ * cache lag during pod creation/update event propagation is normal; only
sustained absence
+ * triggers an apiserver lookup. Set to 0 to verify on every cache miss.
+ */
+ public static final ConfigOption<Integer>
MISSING_DRIVER_GRACE_PERIOD_SECONDS =
+ ConfigOption.<Integer>builder()
+
.key("spark.kubernetes.operator.reconciler.missingDriverGracePeriodSeconds")
+ .enableDynamicOverride(true)
+ .description(
+ "Grace period (in seconds) the operator waits for the pod
informer cache to "
+ + "catch up after the driver pod is observed missing, before
doing a live "
+ + "API verification against the Kubernetes apiserver. This
protects healthy "
+ + "applications from being failed when the informer cache is
briefly stale "
+ + "(e.g., after a watch reconnect). Set to 0 to verify on
every cache miss.")
+ .typeParameterClass(Integer.class)
+ .defaultValue(60)
+ .build();
+
+ /**
+ * Requeue interval (in seconds) used to re-check a driver pod that is
missing from the informer
+ * cache, while waiting for the cache to catch up during the missing-driver
grace period. This is
+ * specific to the missing-driver flow and is not the operator's default
requeue cadence; the
+ * steady-state periodic reconcile interval is {@link
#RECONCILER_INTERVAL_SECONDS}. Set to 0 to
+ * requeue immediately (no delay). Values below 0 are treated as 0.
+ */
+ public static final ConfigOption<Integer>
MISSING_DRIVER_REQUEUE_INTERVAL_SECONDS =
+ ConfigOption.<Integer>builder()
+
.key("spark.kubernetes.operator.reconciler.missingDriverRequeueIntervalSeconds")
+ .enableDynamicOverride(true)
+ .description(
+ "Requeue interval (in seconds) used to re-check a driver pod
that is missing from "
+ + "the informer cache, while waiting for the cache to catch
up during the "
+ + "missing-driver grace period. Specific to the
missing-driver flow; this is "
+ + "not the operator's default requeue cadence. The
steady-state periodic "
+ + "reconcile interval is
spark.kubernetes.operator.reconciler.intervalSeconds. "
+ + "Set to 0 to requeue immediately (no delay); values below
0 are treated as 0.")
+ .typeParameterClass(Integer.class)
+ .defaultValue(15)
+ .build();
+
/**
* When enabled, operator would trim state transition history when a new
attempt starts, keeping
* previous attempt summary only.
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java
index 30d0e4a..55c9ed0 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/context/SparkAppContext.java
@@ -53,6 +53,9 @@ public class SparkAppContext extends
BaseContext<SparkApplication> {
/** secondaryResourceSpec is initialized in a lazy fashion - built upon the
first attempt */
private SparkAppResourceSpec secondaryResourceSpec;
+ /** driverPodFromApi is initialized in a lazy fashion - built upon the first
attempt */
+ private Optional<Pod> driverPodFromApi;
+
/**
* Returns the driver pod for the Spark application, if present.
*
@@ -70,6 +73,30 @@ public class SparkAppContext extends
BaseContext<SparkApplication> {
.findAny();
}
+ /**
+ * Live lookup of the driver pod against the API server, bypassing the
informer cache. The result
+ * is memoized for the lifetime of this context so that multiple observe
steps share a single
+ * API call.
+ * @return An Optional containing the driver Pod as known to the API server.
+ */
+ public Optional<Pod> getDriverPodFromApi() {
+ synchronized (this) {
+ if (driverPodFromApi == null) {
+ driverPodFromApi =
+ josdkContext
+ .getClient()
+ .pods()
+ .inNamespace(sparkApplication.getMetadata().getNamespace())
+ .withLabels(driverLabels(sparkApplication))
+ .list()
+ .getItems()
+ .stream()
+ .findAny();
+ }
+ return driverPodFromApi;
+ }
+ }
+
/**
* Returns a set of executor pods for the Spark application.
*
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 5ab207c..531800d 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
@@ -24,13 +24,16 @@ import static
org.apache.spark.k8s.operator.reconciler.ReconcileProgress.proceed
import static
org.apache.spark.k8s.operator.utils.SparkAppStatusUtils.driverUnexpectedRemoved;
import java.time.Duration;
+import java.time.Instant;
import java.util.List;
import java.util.Optional;
import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.client.KubernetesClientException;
import lombok.extern.log4j.Log4j2;
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.reconciler.ReconcileProgress;
import
org.apache.spark.k8s.operator.reconciler.observers.BaseAppDriverObserver;
@@ -65,30 +68,70 @@ public abstract sealed class AppReconcileStep
final SparkAppContext context,
final SparkAppStatusRecorder statusRecorder,
final List<BaseAppDriverObserver> observers) {
- Optional<Pod> driverPodOptional = context.getDriverPod();
SparkApplication app = context.getResource();
ApplicationStatus currentStatus = app.getStatus();
- if (driverPodOptional.isPresent()) {
- List<ApplicationState> stateUpdates =
- observers.stream()
- .map(o -> o.observe(driverPodOptional.get(), app.getSpec(),
app.getStatus()))
- .filter(Optional::isPresent)
- .map(Optional::get)
- .toList();
- if (stateUpdates.isEmpty()) {
- return proceed();
- } else {
- for (ApplicationState state : stateUpdates) {
- currentStatus = currentStatus.appendNewState(state);
+ Optional<Pod> driverPodOptional = context.getDriverPod();
+
+ if (driverPodOptional.isEmpty()) {
+ Duration verifyAfter =
+
Duration.ofSeconds(SparkOperatorConf.MISSING_DRIVER_GRACE_PERIOD_SECONDS.getValue());
+ Duration requeueAfter =
+ Duration.ofSeconds(
+ Math.max(0,
SparkOperatorConf.MISSING_DRIVER_REQUEUE_INTERVAL_SECONDS.getValue()));
+ ApplicationState observedState = currentStatus.getCurrentState();
+ Instant transitionedAt =
Instant.parse(observedState.getLastTransitionTime());
+ Duration stateAge = Duration.between(transitionedAt, Instant.now());
+
+ if (stateAge.compareTo(verifyAfter) < 0) {
+ log.debug(
+ "Driver pod missing from informer cache; state {} is only {}s old,
"
+ + "deferring verification.",
+ observedState.getCurrentStateSummary(), stateAge.toSeconds());
+ return ReconcileProgress.completeAndRequeueAfter(requeueAfter);
+ }
+
+ try {
+ Optional<Pod> liveDriver = context.getDriverPodFromApi();
+ if (liveDriver.isPresent()) {
+ log.warn(
+ "Driver pod {} missing from informer cache after {}s in state {}
but "
+ + "present on apiserver; informer is likely stale. Using
live pod "
+ + "for this reconcile.",
+ liveDriver.get().getMetadata().getName(),
+ stateAge.toSeconds(),
+ observedState.getCurrentStateSummary());
+ driverPodOptional = liveDriver;
+ } else {
+ ApplicationStatus updatedStatus =
+ currentStatus.appendNewState(driverUnexpectedRemoved());
+ return attemptStatusUpdate(
+ context, statusRecorder, updatedStatus,
completeAndImmediateRequeue());
}
- return attemptStatusUpdate(
- context, statusRecorder, currentStatus,
completeAndImmediateRequeue());
+ } catch (KubernetesClientException e) {
+ log.error(
+ "Driver pod missing from informer cache and live API verification
failed "
+ + "after {}s; deferring removal verdict and requeuing.",
+ stateAge.toSeconds(),
+ e);
+ return ReconcileProgress.completeAndRequeueAfter(requeueAfter);
}
- } else {
- ApplicationStatus updatedStatus =
currentStatus.appendNewState(driverUnexpectedRemoved());
- return attemptStatusUpdate(
- context, statusRecorder, updatedStatus,
completeAndImmediateRequeue());
}
+
+ Pod driverPod = driverPodOptional.get();
+ List<ApplicationState> stateUpdates =
+ observers.stream()
+ .map(o -> o.observe(driverPod, app.getSpec(), app.getStatus()))
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .toList();
+ if (stateUpdates.isEmpty()) {
+ return proceed();
+ }
+ for (ApplicationState state : stateUpdates) {
+ currentStatus = currentStatus.appendNewState(state);
+ }
+ return attemptStatusUpdate(
+ context, statusRecorder, currentStatus, completeAndImmediateRequeue());
}
/**
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppDriverPodCacheStaleTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppDriverPodCacheStaleTest.java
new file mode 100644
index 0000000..6751b78
--- /dev/null
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppDriverPodCacheStaleTest.java
@@ -0,0 +1,169 @@
+/*
+ * 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.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+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.stream.Stream;
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.apache.spark.k8s.operator.Constants;
+import org.apache.spark.k8s.operator.SparkAppSubmissionWorker;
+import org.apache.spark.k8s.operator.SparkApplication;
+import org.apache.spark.k8s.operator.context.SparkAppContext;
+import
org.apache.spark.k8s.operator.reconciler.observers.AppDriverStartObserver;
+import org.apache.spark.k8s.operator.spec.ApplicationSpec;
+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)
+@SuppressFBWarnings(
+ value = {"UWF_UNWRITTEN_FIELD", "NP_UNWRITTEN_FIELD", "UUF_UNUSED_FIELD"},
+ justification = "Fields are written by the Kubernetes mock client
extension.")
+class AppDriverPodCacheStaleTest {
+ private static final String APP_NAME = "stale-cache-app";
+ private static final String NAMESPACE = "default";
+ private static final String OPERATOR_NAME = "spark-kubernetes-operator";
+ private static final String DRIVER_POD_NAME = "stale-cache-app-driver";
+
+ private KubernetesMockServer mockServer;
+ private KubernetesClient kubernetesClient;
+
+ private SparkApplication app;
+ private SparkAppContext context;
+ private SparkAppStatusRecorder recorder;
+
+ @BeforeEach
+ void setUp() {
+ app = new SparkApplication();
+ app.setMetadata(
+ new
ObjectMetaBuilder().withName(APP_NAME).withNamespace(NAMESPACE).build());
+ app.setSpec(new ApplicationSpec());
+ app.setStatus(new ApplicationStatus());
+
+ @SuppressWarnings("unchecked")
+ Context<SparkApplication> josdkContext = mock(Context.class);
+ when(josdkContext.getClient()).thenReturn(kubernetesClient);
+ // Simulate a stale informer: secondary resource cache returns no pods
even though
+ // the apiserver may have the driver pod.
+
when(josdkContext.getSecondaryResourcesAsStream(Pod.class)).thenReturn(Stream.empty());
+
+ context = new SparkAppContext(app, josdkContext,
mock(SparkAppSubmissionWorker.class));
+
+ recorder = mock(SparkAppStatusRecorder.class);
+ when(recorder.persistStatus(any(SparkAppContext.class),
any(ApplicationStatus.class)))
+ .thenAnswer(
+ invocation -> {
+ app.setStatus(invocation.getArgument(1));
+ return true;
+ });
+ }
+
+ @Test
+ void staleCacheWithHealthyDriverOnApiServerDoesNotFailTheApplication() {
+ createDriverPodOnApiServer();
+ setCurrentStateAged(ApplicationStateSummary.DriverRequested,
Duration.ofMinutes(2));
+
+ AppResourceObserveStep step =
+ new AppResourceObserveStep(List.of(new AppDriverStartObserver()));
+ step.reconcile(context, recorder);
+
+ ApplicationStateSummary post =
app.getStatus().getCurrentState().getCurrentStateSummary();
+ assertNotEquals(
+ ApplicationStateSummary.Failed,
+ post,
+ "Stale informer cache must not cause the SparkApp to fail when the
driver "
+ + "pod is healthy on the apiserver.");
+ }
+
+ @Test
+ void youngStateWithCacheMissDefersVerification() {
+ setCurrentStateAged(ApplicationStateSummary.DriverRequested,
Duration.ofSeconds(5));
+
+ AppResourceObserveStep step =
+ new AppResourceObserveStep(List.of(new AppDriverStartObserver()));
+ step.reconcile(context, recorder);
+
+ assertEquals(
+ ApplicationStateSummary.DriverRequested,
+ app.getStatus().getCurrentState().getCurrentStateSummary(),
+ "Cache miss within grace period must not advance state.");
+ }
+
+ @Test
+ void agedStateWithCacheAndApiServerEmptyTransitionsToFailed() {
+ setCurrentStateAged(ApplicationStateSummary.DriverRequested,
Duration.ofMinutes(2));
+
+ AppResourceObserveStep step =
+ new AppResourceObserveStep(List.of(new AppDriverStartObserver()));
+ step.reconcile(context, recorder);
+
+ assertEquals(
+ ApplicationStateSummary.Failed,
+ app.getStatus().getCurrentState().getCurrentStateSummary(),
+ "Genuine driver removal (cache and apiserver agree pod is gone) must
still "
+ + "transition the SparkApp to Failed.");
+ }
+
+ private void createDriverPodOnApiServer() {
+ Pod driverPod =
+ new PodBuilder()
+ .withNewMetadata()
+ .withName(DRIVER_POD_NAME)
+ .withNamespace(NAMESPACE)
+ .withLabels(
+ Map.of(
+ Constants.LABEL_SPARK_OPERATOR_NAME, OPERATOR_NAME,
+ Constants.LABEL_SPARK_APPLICATION_NAME, APP_NAME,
+ Constants.LABEL_SPARK_ROLE_NAME,
Constants.LABEL_SPARK_ROLE_DRIVER_VALUE))
+ .endMetadata()
+ .withNewStatus()
+ .withPhase("Running")
+ .endStatus()
+ .build();
+
kubernetesClient.pods().inNamespace(NAMESPACE).resource(driverPod).create();
+ }
+
+ private void setCurrentStateAged(ApplicationStateSummary summary, Duration
age) {
+ ApplicationState state = new ApplicationState(summary, summary.name());
+ state.setLastTransitionTime(Instant.now().minus(age).toString());
+ app.setStatus(app.getStatus().appendNewState(state));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
