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 54b25b6678e4 [SPARK-60079][K8S][TEST] Add `ExecutorResizePlugin`
integration test
54b25b6678e4 is described below
commit 54b25b6678e4abe2b74605b2f64bad763777cdb2
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Thu Oct 8 05:02:07 2026 -0700
[SPARK-60079][K8S][TEST] Add `ExecutorResizePlugin` integration test
### What changes were proposed in this pull request?
This PR aims to add an integration test for `ExecutorResizePlugin`.
The test runs `JavaSparkSQLCli` with `SELECT reflect('java.lang.Thread',
'sleep', 600000L)` so that the executor stays alive while the plugin works.
The plugin uses the following configurations:
- `spark.kubernetes.executor.resizeInterval=5s`
- `spark.kubernetes.executor.resizeThreshold=0.01`, so that any running
executor exceeds it.
- `spark.kubernetes.executor.resizeFactor=0.1`
- `spark.kubernetes.executor.resizeMaxMemory=1536m`, which is lower than
the first increased limit (`1408Mi * 1.1`). So, the first resize reaches the
maximum.
The test verifies the following:
- The driver log shows that executor 1 is resized to `1536Mi`. A later log
line shows that the next resize is skipped because the maximum was reached.
- The executor pod spec has `1536Mi` as both the memory request and the
memory limit.
- The container status reports `1536Mi` as the memory limit. This means the
kubelet applied the resize in place.
- The container restart count is `0`.
The test is skipped when the `metrics.k8s.io` API is unavailable. It is
also skipped in the `KubernetesSuite` subclasses like `VolcanoSuite`, because
the resize does not depend on the scheduler.
### Why are the changes needed?
`ExecutorResizePlugin` has only unit tests with a mocked
`KubernetesClient`. This PR adds a real K8s cluster test to cover the following:
- Reading memory usage through the `metrics.k8s.io` API.
- Patching the `resize` subresource.
- The in-place resize by the kubelet.
This will provide the test coverage.
- https://github.com/apache/spark/pull/54215
- https://github.com/apache/spark/pull/58582
- https://github.com/apache/spark/pull/58711
- https://github.com/apache/spark/pull/59193
- https://github.com/apache/spark/pull/59281
### Does this PR introduce _any_ user-facing change?
No. This is a test-only change.
### How was this patch tested?
Pass the CIs.
```
[info] - SPARK-60079: Resize executor memory in place with
ExecutorResizePlugin (29 seconds, 130 milliseconds)
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59289 from dongjoon-hyun/SPARK-60079.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../k8s/integrationtest/ExecutorResizeSuite.scala | 87 ++++++++++++++++++++++
.../k8s/integrationtest/KubernetesSuite.scala | 3 +-
2 files changed, 89 insertions(+), 1 deletion(-)
diff --git
a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/ExecutorResizeSuite.scala
b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/ExecutorResizeSuite.scala
new file mode 100644
index 000000000000..326e7418b76d
--- /dev/null
+++
b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/ExecutorResizeSuite.scala
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.deploy.k8s.integrationtest
+
+import scala.jdk.CollectionConverters._
+
+import io.fabric8.kubernetes.api.model.Quantity
+import org.scalatest.concurrent.Eventually
+
+import org.apache.spark.internal.config.PLUGINS
+
+private[spark] trait ExecutorResizeSuite { k8sSuite: KubernetesSuite =>
+
+ import ExecutorResizeSuite._
+ import KubernetesSuite.{k8sTestTag, INTERVAL, SPARK_SQL_CLI_MAIN_CLASS,
TIMEOUT}
+
+ test("SPARK-60079: Resize executor memory in place with
ExecutorResizePlugin", k8sTestTag) {
+ // The resize does not depend on the scheduler, so skip the reruns in
subclasses.
+ assume(this.getClass.getSimpleName == "KubernetesSuite")
+
assume(kubernetesTestComponents.kubernetesClient.hasApiGroup("metrics.k8s.io",
true),
+ "ExecutorResizePlugin requires the metrics.k8s.io API")
+ sparkAppConf
+ .set(PLUGINS.key, "ExecutorResizePlugin")
+ .set("spark.kubernetes.executor.resizeInterval", "5s")
+ // Even an idle executor exceeds this threshold.
+ .set("spark.kubernetes.executor.resizeThreshold", "0.01")
+ .set("spark.kubernetes.executor.resizeFactor", "0.1")
+ // Lower than the first increased limit, so the first resize reaches the
maximum.
+ .set("spark.kubernetes.executor.resizeMaxMemory", "1536m")
+
+ runSparkApplicationAndVerifyCompletion(
+ appResource = containerLocalSparkDistroExamplesJar,
+ mainClass = SPARK_SQL_CLI_MAIN_CLASS,
+ expectedDriverLogOnCompletion = Seq(
+ "Initialized driver component for plugin " +
+ "org.apache.spark.scheduler.cluster.k8s.ExecutorResizePlugin",
+ "Increase executor 1 container memory from",
+ s"to $MAX_MEMORY as usage",
+ s"Skip resizing executor 1 as container memory limit $MAX_MEMORY
already reached " +
+ s"the maximum $MAX_MEMORY"),
+ // Keep the application running until the plugin resizes the executor.
+ appArgs = Array("SELECT reflect('java.lang.Thread', 'sleep', 600000L)"),
+ driverPodChecker = doBasicDriverPodCheck,
+ executorPodChecker = doBasicExecutorPodCheck,
+ isJVM = true)
+
+ // The application is still running, so the executor pod is alive.
+ val pods = kubernetesTestComponents.kubernetesClient
+ .pods()
+ .inNamespace(kubernetesTestComponents.namespace)
+ .withLabel("spark-app-locator", appLocator)
+ .withLabel("spark-role", "executor")
+ .withLabel("spark-exec-id", "1")
+ Eventually.eventually(TIMEOUT, INTERVAL) {
+ val pod = pods.list().getItems.get(0)
+ val container = pod.getSpec.getContainers.asScala
+ .find(_.getName == "spark-kubernetes-executor").get
+ assert(toBytes(container.getResources.getLimits.get("memory")) ===
MAX_MEMORY)
+ assert(toBytes(container.getResources.getRequests.get("memory")) ===
MAX_MEMORY)
+ // The kubelet applies the new limit without restarting the container.
+ val status = pod.getStatus.getContainerStatuses.asScala
+ .find(_.getName == "spark-kubernetes-executor").get
+ assert(toBytes(status.getResources.getLimits.get("memory")) ===
MAX_MEMORY)
+ assert(status.getRestartCount === 0)
+ }
+ }
+
+ private def toBytes(quantity: Quantity): Long =
Quantity.getAmountInBytes(quantity).longValue()
+}
+
+private[spark] object ExecutorResizeSuite {
+ val MAX_MEMORY: Long = 1536L * 1024 * 1024
+}
diff --git
a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
index 624d5b2dae2c..afc769931f09 100644
---
a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
+++
b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala
@@ -44,7 +44,7 @@ class KubernetesSuite extends SparkFunSuite
with BeforeAndAfterAll with BeforeAndAfter with BasicTestsSuite with
SparkConfPropagateSuite
with SecretsTestsSuite with PythonTestsSuite with ClientModeTestsSuite with
PodTemplateSuite
with VolumeSuite with PVTestsSuite with DepsTestsSuite with
DecommissionSuite with RTestsSuite
- with Logging with Eventually with Matchers {
+ with ExecutorResizeSuite with Logging with Eventually with Matchers {
import KubernetesSuite._
@@ -658,6 +658,7 @@ private[spark] object KubernetesSuite {
val SPARK_MINI_READ_WRITE_TEST =
"org.apache.spark.examples.MiniReadWriteTest"
val SPARK_REMOTE_MAIN_CLASS: String =
"org.apache.spark.examples.SparkRemoteFileTest"
val SPARK_DRIVER_MAIN_CLASS: String =
"org.apache.spark.examples.DriverSubmissionTest"
+ val SPARK_SQL_CLI_MAIN_CLASS: String =
"org.apache.spark.examples.sql.JavaSparkSQLCli"
val TIMEOUT = PatienceConfiguration.Timeout(Span(3, Minutes))
val INTERVAL = PatienceConfiguration.Interval(Span(1, Seconds))
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
