This is an automated email from the ASF dual-hosted git repository.
dongjoon 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 096e3a31cd05 [SPARK-54188][K8S] Improve
`ExecutorPodsWatchSnapshotSource` to watch only active executors
096e3a31cd05 is described below
commit 096e3a31cd054fea1b2dfcaff9732a8edc63dc72
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Nov 4 20:56:11 2025 -0800
[SPARK-54188][K8S] Improve `ExecutorPodsWatchSnapshotSource` to watch only
active executors
### What changes were proposed in this pull request?
This PR aims to improve `ExecutorPodsWatchSnapshotSource` to watch only
active executors like `ExecutorPodsPollingSnapshotSource`.
### Why are the changes needed?
`ExecutorPodsPollingSnapshotSource` has been monitoring with
`.withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")` condition. We had better
make `ExecutorPodsWatchSnapshotSource` behave consistently.
https://github.com/apache/spark/blob/a8e35c407bc5340f83b35e5a2f0b0767c6baadb0/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsPollingSnapshotSource.scala#L77-L81
### Does this PR introduce _any_ user-facing change?
No. This will reduce Apache Spark's operation overhead because we ignore
inactive pods.
### How was this patch tested?
Pass the CIs with the revised unit test.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #52888 from dongjoon-hyun/SPARK-54188.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit a2f0c8375ab1271e4920f018af998a884384ed16)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSource.scala | 1 +
.../cluster/k8s/ExecutorPodsWatchSnapshotSourceSuite.scala | 7 ++++++-
2 files changed, 7 insertions(+), 1 deletion(-)
diff --git
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSource.scala
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSource.scala
index 6953ed789f79..0d9f19ee11b7 100644
---
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSource.scala
+++
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSource.scala
@@ -64,6 +64,7 @@ class ExecutorPodsWatchSnapshotSource(
.inNamespace(namespace)
.withLabel(SPARK_APP_ID_LABEL, applicationId)
.withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE)
+ .withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")
.watch(new ExecutorPodsWatcher())
}
}
diff --git
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSourceSuite.scala
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSourceSuite.scala
index 61080268cde6..f830abc0d129 100644
---
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSourceSuite.scala
+++
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSourceSuite.scala
@@ -50,6 +50,9 @@ class ExecutorPodsWatchSnapshotSourceSuite extends
SparkFunSuite with BeforeAndA
@Mock
private var executorRoleLabeledPods: LABELED_PODS = _
+ @Mock
+ private var executorRoleLabeledActivePods: LABELED_PODS = _
+
@Mock
private var watchConnection: Watch = _
@@ -66,7 +69,9 @@ class ExecutorPodsWatchSnapshotSourceSuite extends
SparkFunSuite with BeforeAndA
.thenReturn(appIdLabeledPods)
when(appIdLabeledPods.withLabel(SPARK_ROLE_LABEL, SPARK_POD_EXECUTOR_ROLE))
.thenReturn(executorRoleLabeledPods)
-
when(executorRoleLabeledPods.watch(watch.capture())).thenReturn(watchConnection)
+ when(executorRoleLabeledPods.withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL,
"true"))
+ .thenReturn(executorRoleLabeledActivePods)
+
when(executorRoleLabeledActivePods.watch(watch.capture())).thenReturn(watchConnection)
}
test("Watch events should be pushed to the snapshots store as snapshot
updates.") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
