(spark) branch branch-4.3 updated: [SPARK-59840][K8S] Fix `ExecutorResizePlugin` to exclude inactive executor pods

[do...@apache.org]

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

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


The following commit(s) were added to refs/heads/branch-4.3 by this push:
     new 8e0bf2c06d4b [SPARK-59840][K8S] Fix `ExecutorResizePlugin` to exclude inactive executor pods
8e0bf2c06d4b is described below

commit 8e0bf2c06d4b998fb365934694e8c0272836d82a
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Sep 29 05:45:43 2026 -0700

    [SPARK-59840][K8S] Fix `ExecutorResizePlugin` to exclude inactive executor pods
    
    ### What changes were proposed in this pull request?
    
    This PR aims to fix `ExecutorResizePlugin` to exclude inactive executor pods like `ExecutorPodsPollingSnapshotSource` and `ExecutorPodsWatchSnapshotSource`.
    
    ### Why are the changes needed?
    
    When `spark.kubernetes.executor.deleteOnTermination` is `false`, terminated executor pods are kept with the `spark-exec-inactive=true` label. However, `ExecutorResizePlugin` still sends metrics API requests for them at every interval, and these requests fail.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No. This only removes the unnecessary K8s API requests for inactive executor pods.
    
    ### How was this patch tested?
    
    Pass the CIs with the newly added test case.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Claude Opus 5.5
    
    Closes #59113 from dongjoon-hyun/SPARK-59840.
    
    Authored-by: Dongjoon Hyun <do...@apache.org>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
    (cherry picked from commit 916fba029c86a1533e8af313cfb96f38b58932bf)
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../spark/scheduler/cluster/k8s/ExecutorResizePlugin.scala     |  1 +
 .../scheduler/cluster/k8s/ExecutorResizePluginSuite.scala      | 10 ++++++++++
 2 files changed, 11 insertions(+)

diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorResizePlugin.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorResizePlugin.scala
index 75113c8ebb66..ad4611b52fc7 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorResizePlugin.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorResizePlugin.scala
@@ -99,6 +99,7 @@ class ExecutorResizeDriverPlugin extends DriverPlugin with Logging {
       .inNamespace(namespace)
       .withLabel(SPARK_APP_ID_LABEL, appId)
       .withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
+      .withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")
       .list()
       .getItems.asScala
 
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorResizePluginSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorResizePluginSuite.scala
index e75a07278f97..e07d93ca107c 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorResizePluginSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorResizePluginSuite.scala
@@ -72,6 +72,7 @@ class ExecutorResizePluginSuite
     when(podOperations.inNamespace(namespace)).thenReturn(podsWithNamespace)
     when(podsWithNamespace.withLabel(SPARK_APP_ID_LABEL, appId)).thenReturn(labeledPods)
     when(labeledPods.withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)).thenReturn(labeledPods)
+    when(labeledPods.withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")).thenReturn(labeledPods)
     when(labeledPods.list()).thenReturn(podList)
     when(kubernetesClient.top()).thenReturn(topOperations)
     when(topOperations.pods()).thenReturn(podMetricOperations)
@@ -147,6 +148,15 @@ class ExecutorResizePluginSuite
     verify(podMetricOperations, never()).metrics(anyString(), anyString())
   }
 
+  test("SPARK-59840: Inactive executor pods are excluded from the listing") {
+    val plugin = createPlugin()
+    when(podList.getItems).thenReturn(Collections.emptyList())
+
+    plugin.invokePrivate(_checkAndIncreaseMemory(namespace, 0.9, 0.1, kubernetesClient))
+
+    verify(labeledPods).withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")
+  }
+
   test("Memory usage below threshold should not trigger resize") {
     val plugin = createPlugin()
     val pod = createPodWithMemoryLimit(1, "1000000000") // 1GB limit


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