(spark) branch master updated: [SPARK-56670][K8S] Restrict `ExecutorResizePlugin` to `direct` pods allocator

[do...@apache.org]

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 4022753b2523 [SPARK-56670][K8S] Restrict `ExecutorResizePlugin` to `direct` pods allocator
4022753b2523 is described below

commit 4022753b2523ec0601eaaa8318a9ae8a02a5f524
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Wed Apr 29 21:26:24 2026 -0700

    [SPARK-56670][K8S] Restrict `ExecutorResizePlugin` to `direct` pods allocator
    
    ### What changes were proposed in this pull request?
    
    This PR restricts `ExecutorResizePlugin` to activate only when the `direct` pods allocator (`ExecutorPodsAllocator`) is configured via `spark.kubernetes.allocation.pods.allocator`.
    
    When the configured allocator is anything else (`statefulset`, `deployment`, or a user-supplied class), `ExecutorResizeDriverPlugin.init()` now logs a warning and returns early without creating a Kubernetes client or
    scheduling the periodic memory check.
    
    ### Why are the changes needed?
    
    `ExecutorResizePlugin` resizes individual executor pods in place via `pods().withName(...).subresource("resize").patch(...)`. This works correctly only when Spark itself owns each pod, i.e. with the `direct` allocator. For the other allocators, user-defined or the higher-level controller's pod template is the source of truth and patching a single pod under those allocators may have unexpected side-effects.
    - `statefulset` → `StatefulSet.spec.template`
    - `deployment`  → `Deployment.spec.template`
    
    ### Does this PR introduce _any_ user-facing change?
    
    No, `ExecutorResizePlugin` is a new unreleased feature of Apache Spark 4.2.0.
    
    ### How was this patch tested?
    
    Pass the CIs with the newly added test coverage.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Claude Code (Opus 4.7)
    
    Closes #55615 from dongjoon-hyun/SPARK-56670.
    
    Authored-by: Dongjoon Hyun <do...@apache.org>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../cluster/k8s/ExecutorResizePlugin.scala          | 10 +++++++++-
 .../cluster/k8s/ExecutorResizePluginSuite.scala     | 21 ++++++++++++++++++++-
 2 files changed, 29 insertions(+), 2 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 77fd37dd538b..d5d882978dfc 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
@@ -32,7 +32,7 @@ import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.deploy.k8s.SparkKubernetesClientFactory
 import org.apache.spark.internal.Logging
-import org.apache.spark.internal.LogKeys.{EXECUTOR_ID, MEMORY_SIZE}
+import org.apache.spark.internal.LogKeys.{CONFIG, CONFIG2, EXECUTOR_ID, MEMORY_SIZE}
 import org.apache.spark.util.{ThreadUtils, Utils}
 
 /**
@@ -53,6 +53,14 @@ class ExecutorResizeDriverPlugin extends DriverPlugin with Logging {
     ThreadUtils.newDaemonSingleThreadScheduledExecutor("executor-resize-plugin")
 
   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; " +
+        log"${MDC(CONFIG, KUBERNETES_ALLOCATION_PODS_ALLOCATOR.key)} is " +
+        log"${MDC(CONFIG2, allocator)}. Plugin will not start.")
+      return Map.empty[String, String].asJava
+    }
+
     val interval = Utils.timeStringAsSeconds(
       sc.conf.get(EXECUTOR_RESIZE_INTERVAL.key, "1m"))
     val threshold = sc.conf.getDouble(EXECUTOR_RESIZE_THRESHOLD.key, 0.9)
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 649ccd8a945d..e8ce7d44abd0 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
@@ -29,7 +29,9 @@ import org.mockito.Mockito.{mock, never, times, verify, when}
 import org.scalatest.BeforeAndAfter
 import org.scalatest.PrivateMethodTester
 
-import org.apache.spark.{SparkContext, SparkFunSuite}
+import org.apache.spark.{SparkConf, SparkContext, SparkFunSuite}
+import org.apache.spark.api.plugin.PluginContext
+import org.apache.spark.deploy.k8s.Config.KUBERNETES_ALLOCATION_PODS_ALLOCATOR
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.deploy.k8s.Fabric8Aliases._
 
@@ -287,4 +289,21 @@ class ExecutorResizePluginSuite
 
     verify(podResource, times(1)).patch(any(), any(classOf[Pod]))
   }
+
+  Seq("statefulset", "deployment").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)
+      val sc = mock(classOf[SparkContext])
+      when(sc.conf).thenReturn(sparkConf)
+      val pluginCtx = mock(classOf[PluginContext])
+
+      val result = plugin.init(sc, pluginCtx)
+
+      assert(result.isEmpty)
+      val clientField = plugin.getClass.getDeclaredField("kubernetesClient")
+      clientField.setAccessible(true)
+      assert(clientField.get(plugin) == null)
+    }
+  }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org