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 913175ed22f6 [SPARK-59918][K8S] Support `deployment` pods allocator in
`ExecutorResizePlugin`
913175ed22f6 is described below
commit 913175ed22f616bb229d9f4533a398f832680cd5
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Sep 30 22:23:15 2026 -0700
[SPARK-59918][K8S] Support `deployment` pods allocator in
`ExecutorResizePlugin`
### What changes were proposed in this pull request?
This PR aims to support `ExecutorResizePlugin` with the `deployment` pods
allocator for Apache Spark 4.4.0 in addition to the default `direct` pods
allocator.
### Why are the changes needed?
[SPARK-56670](https://github.com/apache/spark/pull/55615) restricted
`ExecutorResizePlugin` to the `direct` pods allocator because the higher-level
controller's pod template is the source of truth for the other allocators with
different mechanism.
However, after SPARK-59914, Spark becomes capable of support `Deployment`
because we never updates its pod template, and the `ReplicaSet` controller
neither reverts in-place pod resizes nor rolls out a new revision for them.
- https://github.com/apache/spark/pull/59175
### Does this PR introduce _any_ user-facing change?
Yes. In Apache Spark 4.2.0, `ExecutorResizePlugin` logs a warning and does
not start with the `deployment` pods allocator. After this PR, it works with
the `deployment` pods allocator.
### How was this patch tested?
Pass the CIs with the updated test case.
I also verified manually on K8s v1.37.0 that the executor pods of the
`Deployment` are resized in place up to
`spark.kubernetes.executor.resizeMaxMemory` without restarts, while the
`Deployment` keeps its pod template and a single `ReplicaSet`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59180 from
dongjoon-hyun/dongjoon/executor-resize-plugin-deployment-0e6321.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
docs/running-on-kubernetes.md | 3 ++-
.../scheduler/cluster/k8s/ExecutorResizePlugin.scala | 4 ++--
.../cluster/k8s/ExecutorResizePluginSuite.scala | 18 +++++++++++++++++-
3 files changed, 21 insertions(+), 4 deletions(-)
diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index dc83f056b70f..ace9686ba876 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -579,7 +579,8 @@ change to the application code. They apply to executor pods
of every resource pr
The resize plugins are registered via `spark.plugins` and run in the driver,
while the recovery mode is
built into the executor pod allocator and needs no plugin. All three features
require the default
-`direct` pods allocator (`spark.kubernetes.allocation.pods.allocator`).
+`direct` pods allocator (`spark.kubernetes.allocation.pods.allocator`), except
that the memory resize
+plugin also supports the `deployment` pods allocator.
### Executor Memory Resize
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 7c2dc44c525a..28d43b2d68b7 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
@@ -56,8 +56,8 @@ class ExecutorResizeDriverPlugin extends DriverPlugin with
Logging {
override def init(sc: SparkContext, ctx: PluginContext): JMap[String,
String] = {
val allocator = sc.conf.get(KUBERNETES_ALLOCATION_PODS_ALLOCATOR)
- if (allocator != "direct") {
- logWarning(log"ExecutorResizePlugin requires the 'direct' pods
allocator; " +
+ if (allocator != "direct" && allocator != "deployment") {
+ logWarning(log"ExecutorResizePlugin requires the 'direct' or
'deployment' pods allocator; " +
log"${MDC(CONFIG, KUBERNETES_ALLOCATION_PODS_ALLOCATOR.key)} is " +
log"${MDC(CONFIG2, allocator)}. Plugin will not start.")
return Map.empty[String, String].asJava
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 32a8ff239e84..6b6e6bea029d 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
@@ -381,7 +381,7 @@ class ExecutorResizePluginSuite
assert(countSkipLogs(2) === 1)
}
- Seq("statefulset", "deployment").foreach { allocator =>
+ Seq("statefulset").foreach { allocator =>
test(s"init returns early when pods allocator is '$allocator'") {
val plugin = new ExecutorResizeDriverPlugin()
val sparkConf = new
SparkConf().set(KUBERNETES_ALLOCATION_PODS_ALLOCATOR, allocator)
@@ -395,6 +395,22 @@ class ExecutorResizePluginSuite
}
}
+ test("SPARK-59918: init schedules the resize task when pods allocator is
'deployment'") {
+ val plugin = new ExecutorResizeDriverPlugin()
+ val sparkConf = new SparkConf().set(KUBERNETES_ALLOCATION_PODS_ALLOCATOR,
"deployment")
+ val sc = mock(classOf[SparkContext])
+ when(sc.conf).thenReturn(sparkConf)
+ val pluginCtx = mock(classOf[PluginContext])
+ try {
+ assert(plugin.init(sc, pluginCtx).isEmpty)
+ val field = plugin.getClass.getDeclaredField("periodicService")
+ field.setAccessible(true)
+
assert(field.get(plugin).asInstanceOf[ScheduledThreadPoolExecutor].getQueue.size
=== 1)
+ } finally {
+ plugin.shutdown()
+ }
+ }
+
Seq(
(EXECUTOR_RESIZE_THRESHOLD, Seq("0", "1", "1.5"), "The threshold should be
in (0, 1)"),
(EXECUTOR_RESIZE_FACTOR, Seq("-0.1", "0", "1.5"), "The factor should be in
(0, 1]")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
