(spark) branch branch-4.1 updated: [SPARK-59008][K8S] Fix NPE in ExecutorPodsLifecycleManager when pod is deleted between get() calls

[do...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dongjoon-hyun pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.1 by this push:
     new 6c0d65852095 [SPARK-59008][K8S] Fix NPE in ExecutorPodsLifecycleManager when pod is deleted between get() calls
6c0d65852095 is described below

commit 6c0d6585209584491f417ee83afb2a812d3d4bc3
Author: Venkata krishnan Sowrirajan <ve...@gmail.com>
AuthorDate: Wed Aug 26 18:16:46 2026 -0700

    [SPARK-59008][K8S] Fix NPE in ExecutorPodsLifecycleManager when pod is deleted between get() calls
    
    ### What changes were proposed in this pull request?
    
    - Fetch the pod once in `ExecutorPodsLifecycleManager.removeExecutorFromK8s` instead of calling `podToDelete.get()` twice.
    
    ### Why are the changes needed?
    
    - `podToDelete` is a lazy `PodResource` handle, so each `get()` is a separate API server round trip. If the pod is deleted between the two calls, the second returns `null` and dereferencing `getMetadata` throws `NullPointerException: Cannot invoke "io.fabric8.kubernetes.api.model.Pod.getMetadata()" because the return value of "io.fabric8.kubernetes.client.dsl.PodResource.get()" is null`, thrown from `removeExecutorFromK8s` via `onFinalNonDeletedState` / `onNewSnapshots`.
    - Observed on a driver managing ~1000 executors during heavy pod churn. The exception aborts the K8s-side deletion for that executor, leaving the pod to be reaped on a later resync.
    - Introduced by SPARK-54197, which added the `deletionTimestamp` check as a second `get()` call rather than reusing the first result. The `&&` short-circuit guards against the first call returning `null`, but not against the pod disappearing between the two.
    
    ### Does this PR introduce _any_ user-facing change?
    
    - No.
    
    ### How was this patch tested?
    
    - New unit test in `ExecutorPodsLifecycleManagerSuite` stubbing `get()` to return the pod then `null` on successive calls. Fails on master with the NPE above, passes with this change. Full suite 10/10.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    - Yes, using Claude with Opus 4.8.
    
    Closes #58291 from venkata91/SPARK-executor-pods-lifecycle-npe.
    
    Authored-by: Venkata krishnan Sowrirajan <ve...@gmail.com>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
    (cherry picked from commit b18e46d4fbdab45bb123efd1de1dbd1a619f1f34)
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../cluster/k8s/ExecutorPodsLifecycleManager.scala       |  6 ++++--
 .../cluster/k8s/ExecutorPodsLifecycleManagerSuite.scala  | 16 ++++++++++++++++
 2 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsLifecycleManager.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsLifecycleManager.scala
index 3a508add6ccf..17db0dae5099 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsLifecycleManager.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsLifecycleManager.scala
@@ -218,8 +218,10 @@ private[spark] class ExecutorPodsLifecycleManager(
           .inNamespace(namespace)
           .withName(updatedPod.getMetadata.getName)
 
-        if (podToDelete.get() != null &&
-            podToDelete.get.getMetadata.getDeletionTimestamp == null) {
+        // Fetch once: the pod can be removed between two `get` calls, making the second
+        // return null and NPE while dereferencing its metadata.
+        val fetchedPod = podToDelete.get()
+        if (fetchedPod != null && fetchedPod.getMetadata.getDeletionTimestamp == null) {
           podToDelete.delete()
         }
       } else if (!inactivatedPods.contains(execId) && !isPodInactive(updatedPod)) {
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsLifecycleManagerSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsLifecycleManagerSuite.scala
index cdbcae050ceb..ab5034aac3d2 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsLifecycleManagerSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsLifecycleManagerSuite.scala
@@ -169,6 +169,22 @@ class ExecutorPodsLifecycleManagerSuite extends SparkFunSuite with BeforeAndAfte
     verify(namedExecutorPods(failedPod.getMetadata.getName), times(1)).delete()
   }
 
+  test("SPARK-59008: Pod deleted between the two get calls doesn't throw NPE.") {
+    val failedPod = failedExecutorWithoutDeletion(1)
+    val mockPodResource = mock(classOf[PodResource])
+    namedExecutorPods.put("spark-executor-1", mockPodResource)
+    // The pod is present on the first lookup and gone right after, as when the API server
+    // removes it concurrently with the driver's deletion attempt. Re-reading it would NPE.
+    when(mockPodResource.get()).thenReturn(failedPod, null.asInstanceOf[Pod])
+    snapshotsStore.updatePod(failedPod)
+    snapshotsStore.notifySubscribers()
+
+    val msg = exitReasonMessage(1, failedPod, 1)
+    val expectedLossReason = ExecutorExited(1, exitCausedByApp = true, msg)
+    verify(schedulerBackend, times(1)).doRemoveExecutor("1", expectedLossReason)
+    verify(namedExecutorPods(failedPod.getMetadata.getName), times(1)).delete()
+  }
+
   test("When the scheduler backend lists executor ids that aren't present in the cluster," +
     " remove those executors from Spark.") {
       when(schedulerBackend.getExecutorsWithRegistrationTs()).thenReturn(Map("1" -> 7L))


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