This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 6ed637f37301 [SPARK-59914][K8S] Make `ExecutorResizePlugin` skip
executor pods with the placeholder `EXECID` label
6ed637f37301 is described below
commit 6ed637f37301b35c578bc79a6d042085d31c9274
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Sep 30 17:48:51 2026 -0700
[SPARK-59914][K8S] Make `ExecutorResizePlugin` skip executor pods with the
placeholder `EXECID` label
### What changes were proposed in this pull request?
This PR aims to make `ExecutorResizePlugin` skip executor pods whose
`spark-exec-id` label is the placeholder `EXECID`, like `ExecutorPodsSnapshot`
does.
### Why are the changes needed?
`deployment` and `statefulset` pods allocators create executor pods with
the `EXECID` label and patch the real executor ID later on a best-effort basis.
Currently, `ExecutorResizePlugin` runs only with the `direct` pods allocator,
so this is a defensive improvement. It keeps the placeholder out of the
`cappedExecutors` keys and the logs.
### Does this PR introduce _any_ user-facing change?
No.
### 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 #59175 from dongjoon-hyun/SPARK-59914.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../scheduler/cluster/k8s/ExecutorResizePlugin.scala | 17 ++++++++++++-----
.../cluster/k8s/ExecutorResizePluginSuite.scala | 18 ++++++++++++++++++
2 files changed, 30 insertions(+), 5 deletions(-)
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 85f59f100837..7c2dc44c525a 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
@@ -111,13 +111,20 @@ class ExecutorResizeDriverPlugin extends DriverPlugin
with Logging {
.list()
.getItems.asScala
+ val executorPods = pods.flatMap { pod =>
+ pod.getMetadata.getLabels.get(SPARK_EXECUTOR_ID_LABEL) match {
+ case "EXECID" | null =>
+ // The exec label has not yet been assigned
+ None
+ case id =>
+ Some((id, pod))
+ }
+ }
+
// Drop executors that no longer exist so that cappedExecutors does not
grow unbounded.
- cappedExecutors.retainAll(pods.flatMap { p =>
- Option(p.getMetadata.getLabels.get(SPARK_EXECUTOR_ID_LABEL))
- }.toSet.asJava)
+ cappedExecutors.retainAll(executorPods.map(_._1).toSet.asJava)
- pods.filter(_.getMetadata.getLabels.get(SPARK_EXECUTOR_ID_LABEL) !=
null).foreach { pod =>
- val execId = pod.getMetadata.getLabels.get(SPARK_EXECUTOR_ID_LABEL)
+ executorPods.foreach { case (execId, pod) =>
try {
val metrics = kubernetesClient.top().pods().metrics(namespace,
pod.getMetadata.getName)
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 ec5cfe48af5b..32a8ff239e84 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
@@ -161,6 +161,24 @@ class ExecutorResizePluginSuite
verify(podMetricOperations, never()).metrics(anyString(), anyString())
}
+ test("Pod with placeholder executor ID label should be skipped") {
+ val plugin = createPlugin()
+ val pod = new PodBuilder()
+ .withNewMetadata()
+ .withName("spark-executor-1")
+ .addToLabels(SPARK_APP_ID_LABEL, appId)
+ .addToLabels(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
+ .addToLabels(SPARK_EXECUTOR_ID_LABEL, "EXECID")
+ .endMetadata()
+ .build()
+
+ when(podList.getItems).thenReturn(Collections.singletonList(pod))
+
+ plugin.invokePrivate(_checkAndIncreaseMemory(namespace, 0.9, 0.1,
kubernetesClient))
+
+ 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())
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
