This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 3b4afd88ef5a [SPARK-59843][K8S] Make `ExecutorResizePlugin` validate
`resizeThreshold` and `resizeFactor`
3b4afd88ef5a is described below
commit 3b4afd88ef5a81ba4066190e0da9e70f33d08122
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Sep 29 05:27:14 2026 -0700
[SPARK-59843][K8S] Make `ExecutorResizePlugin` validate `resizeThreshold`
and `resizeFactor`
This PR aims to make `ExecutorResizePlugin` read
`spark.kubernetes.executor.resizeThreshold` and
`spark.kubernetes.executor.resizeFactor` via their `ConfigEntry`s, like
`ExecutorPVCResizePlugin` does.
Currently, the raw key reads bypass `checkValue`, so invalid values are
silently accepted. For example, `resizeFactor=0` patches the same memory limit
repeatedly, a negative `resizeFactor` shrinks it, and `resizeThreshold>=1`
never triggers a resize.
Yes, but only for invalid configurations. They now fail at driver startup
with the config's validation error instead of being silently accepted.
Pass the CIs with the newly added test cases.
Generated-by: Claude Opus 5.5
Closes #59115 from dongjoon-hyun/SPARK-59843.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit bc1642ad9814fce87416caca3ee4abd554f0c9ca)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
docs/running-on-kubernetes.md | 4 ++--
.../scala/org/apache/spark/deploy/k8s/Config.scala | 4 ++--
.../cluster/k8s/ExecutorResizePlugin.scala | 4 ++--
.../cluster/k8s/ExecutorResizePluginSuite.scala | 27 ++++++++++++++++++++--
4 files changed, 31 insertions(+), 8 deletions(-)
diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index ec613958df04..8c3039e07eb1 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -1989,7 +1989,7 @@ See the [configuration page](configuration.html) for
information on Spark config
<td><code>spark.kubernetes.executor.resizeThreshold</code></td>
<td><code>0.9</code></td>
<td>
- The threshold to resize.
+ The threshold to resize. It should be in (0, 1).
Takes effect only when
<code>org.apache.spark.scheduler.cluster.k8s.ExecutorResizePlugin</code>
is registered via <code>spark.plugins</code>.
</td>
@@ -1999,7 +1999,7 @@ See the [configuration page](configuration.html) for
information on Spark config
<td><code>spark.kubernetes.executor.resizeFactor</code></td>
<td><code>0.1</code></td>
<td>
- The factor to resize.
+ The factor to resize. It should be in (0, 1].
Takes effect only when
<code>org.apache.spark.scheduler.cluster.k8s.ExecutorResizePlugin</code>
is registered via <code>spark.plugins</code>.
</td>
diff --git
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
index 60d61028476f..160f64c77f04 100644
---
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
+++
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
@@ -292,7 +292,7 @@ private[spark] object Config extends Logging {
val EXECUTOR_RESIZE_THRESHOLD =
ConfigBuilder("spark.kubernetes.executor.resizeThreshold")
- .doc("The threshold to resize.")
+ .doc("The threshold to resize. It should be in (0, 1).")
.version("4.2.0")
.doubleConf
.checkValue(v => 0 < v && v < 1, "The threshold should be in (0, 1)")
@@ -300,7 +300,7 @@ private[spark] object Config extends Logging {
val EXECUTOR_RESIZE_FACTOR =
ConfigBuilder("spark.kubernetes.executor.resizeFactor")
- .doc("The factor to resize.")
+ .doc("The factor to resize. It should be in (0, 1].")
.version("4.2.0")
.doubleConf
.checkValue(v => 0 < v && v <= 1, "The factor should be in (0, 1]")
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 5b526db122e3..75113c8ebb66 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
@@ -61,8 +61,8 @@ class ExecutorResizeDriverPlugin extends DriverPlugin with
Logging {
val interval = Utils.timeStringAsSeconds(
sc.conf.get(EXECUTOR_RESIZE_INTERVAL.key, "1m"))
- val threshold = sc.conf.getDouble(EXECUTOR_RESIZE_THRESHOLD.key, 0.9)
- val factor = sc.conf.getDouble(EXECUTOR_RESIZE_FACTOR.key, 0.1)
+ val threshold = sc.conf.get(EXECUTOR_RESIZE_THRESHOLD)
+ val factor = sc.conf.get(EXECUTOR_RESIZE_FACTOR)
val namespace = sc.conf.get(KUBERNETES_NAMESPACE)
// Scheduler is not created yet at init time; resolve it lazily in the
periodic task.
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 9a0439b91f45..e75a07278f97 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,9 +29,9 @@ import org.mockito.Mockito.{mock, never, times, verify, when}
import org.scalatest.BeforeAndAfter
import org.scalatest.PrivateMethodTester
-import org.apache.spark.{SparkConf, SparkContext, SparkFunSuite}
+import org.apache.spark.{SparkConf, SparkContext, SparkFunSuite,
SparkIllegalArgumentException}
import org.apache.spark.api.plugin.PluginContext
-import org.apache.spark.deploy.k8s.Config.KUBERNETES_ALLOCATION_PODS_ALLOCATOR
+import org.apache.spark.deploy.k8s.Config._
import org.apache.spark.deploy.k8s.Constants._
import org.apache.spark.deploy.k8s.Fabric8Aliases._
@@ -301,4 +301,27 @@ class ExecutorResizePluginSuite
assert(result.isEmpty)
}
}
+
+ 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]")
+ ).foreach { case (entry, invalidValues, requirement) =>
+ test(s"SPARK-59843: init fails on invalid ${entry.key}") {
+ invalidValues.foreach { value =>
+ val plugin = new ExecutorResizeDriverPlugin()
+ val sparkConf = new SparkConf().set(entry.key, value)
+ val sc = mock(classOf[SparkContext])
+ when(sc.conf).thenReturn(sparkConf)
+ val pluginCtx = mock(classOf[PluginContext])
+
+ checkError(
+ exception = intercept[SparkIllegalArgumentException](plugin.init(sc,
pluginCtx)),
+ condition = "INVALID_CONF_VALUE.REQUIREMENT",
+ parameters = Map(
+ "confName" -> entry.key,
+ "confValue" -> value.toDouble.toString,
+ "confRequirement" -> requirement))
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
