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 0ccd72732e2d [SPARK-59839][K8S] Prune PVC usage reports of removed and
inactive executors in `ExecutorPVCResizePlugin`
0ccd72732e2d is described below
commit 0ccd72732e2d2c98b52a24c7503a97274e35bcba
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Sep 29 05:53:05 2026 -0700
[SPARK-59839][K8S] Prune PVC usage reports of removed and inactive
executors in `ExecutorPVCResizePlugin`
### What changes were proposed in this pull request?
This PR aims to prune the PVC usage reports of removed and inactive
executors in `ExecutorPVCResizePlugin`.
- Exclude inactive executor pods (`spark-exec-inactive=true`) from the pod
listing, like `ExecutorPodsPollingSnapshotSource` and
`ExecutorPodsWatchSnapshotSource`.
- Drop the reports of executors without a listed pod, like
`ExecutorResizePlugin` does for `cappedExecutors`.
Note that only `latestReports` grows per executor. The PVC-name-keyed maps
(`failedPvcs`, `cappedPvcs`, and `requestedSizes`) are intentionally out of
scope because PVCs can outlive executors via PVC reuse, and they are neither
iterated nor logged.
### Why are the changes needed?
`latestReports` is never pruned. So, in long-running applications with
dynamic allocation, the reports of removed executors accumulate forever and the
whole map is logged at INFO level every interval.
In addition, with `spark.kubernetes.executor.deleteOnTermination=false`,
the stale reports of terminated executor pods keep being applied to their PVCs,
which can be reused by new executors.
### Does this PR introduce _any_ user-facing change?
Yes, previously it was never pruned. This is a bug fix for
`ExecutorPVCResizePlugin`, which was released in
[v4.2.0](https://github.com/apache/spark/releases/tag/v4.2.0) (2026-07-11).
### How was this patch tested?
Pass the CIs with the newly added test cases.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59112 from dongjoon-hyun/SPARK-59839.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../cluster/k8s/ExecutorPVCResizePlugin.scala | 9 +++++++--
.../cluster/k8s/ExecutorPVCResizePluginSuite.scala | 23 ++++++++++++++++++++++
2 files changed, 30 insertions(+), 2 deletions(-)
diff --git
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePlugin.scala
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePlugin.scala
index 2b61de9e645b..43922e1c0dfe 100644
---
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePlugin.scala
+++
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePlugin.scala
@@ -67,7 +67,7 @@ class ExecutorPVCResizeDriverPlugin extends DriverPlugin with
Logging {
private var factor: Double = _
private var maxStorage: Long = Long.MaxValue
- private val latestReports = new ConcurrentHashMap[String,
PVCDiskUsageReport]()
+ private[k8s] val latestReports = new ConcurrentHashMap[String,
PVCDiskUsageReport]()
private val failedPvcs = ConcurrentHashMap.newKeySet[String]()
// PVCs whose storage request already reached maxStorage, to log the skip
only once.
private val cappedPvcs = ConcurrentHashMap.newKeySet[String]()
@@ -121,16 +121,17 @@ class ExecutorPVCResizeDriverPlugin extends DriverPlugin
with Logging {
}
private[k8s] def checkAndResizePVCs(): Unit = {
- logInfo(s"Latest PVC usage reports: $latestReports")
val appId = sparkContext.applicationId
sparkContext.schedulerBackend match {
case b: KubernetesClusterSchedulerBackend =>
val client = b.kubernetesClient
+ // Skip terminated pods kept by deleteOnTermination=false since their
reports are stale.
val pods = client.pods()
.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
@@ -138,6 +139,10 @@ class ExecutorPVCResizeDriverPlugin extends DriverPlugin
with Logging {
Option(p.getMetadata.getLabels.get(SPARK_EXECUTOR_ID_LABEL)).map(_
-> p)
}.toMap
+ // Drop reports of executors without a pod so that latestReports does
not grow unbounded.
+ latestReports.keySet().retainAll(podByExecId.keySet.asJava)
+ logInfo(s"Latest PVC usage reports: $latestReports")
+
latestReports.values().asScala.foreach { report =>
podByExecId.get(report.executorId).foreach { pod =>
pvcsOf(pod).foreach { pvcName =>
diff --git
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePluginSuite.scala
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePluginSuite.scala
index 55fb843016ba..f446d24c3d92 100644
---
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePluginSuite.scala
+++
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPVCResizePluginSuite.scala
@@ -66,6 +66,7 @@ class ExecutorPVCResizePluginSuite
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.persistentVolumeClaims()).thenReturn(pvcOperations)
when(pvcOperations.inNamespace(namespace)).thenReturn(pvcsWithNamespace)
@@ -308,6 +309,28 @@ class ExecutorPVCResizePluginSuite
verify(pvcsWithNamespace,
never()).withName(org.mockito.ArgumentMatchers.anyString())
}
+ test("SPARK-59839: Reports of executors without a pod are removed") {
+ val plugin = createPlugin()
+ val pod = createPodWithPVC(1, "pvc-1", "/data")
+ when(podList.getItems).thenReturn(Collections.singletonList(pod))
+ plugin.receive(PVCDiskUsageReport("1", 0.5))
+ plugin.receive(PVCDiskUsageReport("2", 0.5)) // No pod for executor 2
+ assert(plugin.latestReports.keySet() === java.util.Set.of("1", "2"))
+
+ plugin.checkAndResizePVCs()
+
+ assert(plugin.latestReports.keySet() === Collections.singleton("1"))
+ }
+
+ test("SPARK-59839: Inactive executor pods are excluded from the listing") {
+ val plugin = createPlugin()
+ when(podList.getItems).thenReturn(Collections.emptyList())
+
+ plugin.checkAndResizePVCs()
+
+ verify(labeledPods).withoutLabel(SPARK_EXECUTOR_INACTIVE_LABEL, "true")
+ }
+
test("pvcsOf returns claim names mounted by the executor container") {
val plugin = createPlugin()
val pod = createPodWithPVC(7, "pvc-7", "/spark-local")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
