This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new bc5f163c26c4 [SPARK-59840][K8S] Fix `ExecutorResizePlugin` to exclude
inactive executor pods
bc5f163c26c4 is described below
commit bc5f163c26c47664f76a3dbee756f18807d8eb45
Author: Dongjoon Hyun <[email protected]>
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 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 916fba029c86a1533e8af313cfb96f38b58932bf)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../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 c97d4e484789..4aeb7874a36e 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
@@ -104,6 +104,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 488ec970d396..4e0331461fdd 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
@@ -73,6 +73,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)
@@ -159,6 +160,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: [email protected]
For additional commands, e-mail: [email protected]
