You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ya...@apache.org on 2026/08/20 03:39:01 UTC
(spark) branch branch-4.0 updated: [SPARK-58857][K8S] Bind the result of Utils.randomize in LocalDirsFeatureStep
This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new e3c47695a77c [SPARK-58857][K8S] Bind the result of Utils.randomize in LocalDirsFeatureStep
e3c47695a77c is described below
commit e3c47695a77cf104789288de4a0f46feda6dd6d3
Author: YangJie <ya...@baidu.com>
AuthorDate: Thu Aug 20 11:35:48 2026 +0800
[SPARK-58857][K8S] Bind the result of Utils.randomize in LocalDirsFeatureStep
### What changes were proposed in this pull request?
Bind the result of `Utils.randomize` in `LocalDirsFeatureStep.configurePod`, so the local dirs resolved from configuration are actually shuffled.
Binding the result also makes two `toImmutableArraySeq` conversions redundant, since `randomize` already returns a `Seq`, which in turn leaves the `ArrayImplicits` import unused.
### Why are the changes needed?
`configurePod` calls `randomize` on the resolved dirs and discards the return value:
```scala
val resolvedLocalDirs = Option(conf.sparkConf.getenv("SPARK_LOCAL_DIRS"))
.orElse(conf.getOption("spark.local.dir"))
.getOrElse(defaultLocalDir)
.split(",")
randomize(resolvedLocalDirs) // result dropped
localDirs = resolvedLocalDirs.toImmutableArraySeq // configured order
```
`Utils.randomize[T](seq: IterableOnce[T]): Seq[T]` is `randomizeInPlace(seq.iterator.toArray).toImmutableArraySeq`. It shuffles a fresh copy and returns it, so the caller observes nothing on the argument. The genuinely mutating variant is `Utils.randomizeInPlace`. The sibling call fifteen lines earlier, for the pod-template branch, binds the result and works.
Both calls arrived in the same commit, 9f9af2a7bf6 (SPARK-39755, "Improve `LocalDirsFeatureStep` to randomize local directories"). An earlier revision of that PR used `randomizeInPlace`; review suggested `Utils.randomize`, and the next revision adopted it at both sites, binding the result at one and dropping it at the other.
So on this branch `SPARK_LOCAL_DIRS`, the emptyDir volume list, and the `spark-local-dir-N` to path pairing come out in configured order, identically for every pod in an application.
One thing this PR deliberately does not claim: there is no measurable I/O or capacity benefit. `DiskBlockManager.getFile` selects a directory by `nonNegativeHash(filename) % localDirs.length`, which already spreads files near-uniformly within each executor whatever order it receives, and in this branch every path is an emptyDir this step creates, so they share one node filesystem (or RAM under `spark.kubernetes.local.dirs.tmpfs`). What changes is that the randomization SPARK-39755 add [...]
If reviewers would rather not have the randomization at all, deleting the call as dead code is a reasonable alternative and I am happy to switch. The argument for fixing rather than deleting is parity with the sibling branch and with SPARK-39755's stated intent.
### Does this PR introduce _any_ user-facing change?
The order of `SPARK_LOCAL_DIRS` and of the generated emptyDir volume mounts now varies between pods instead of matching the configured order. No API or configuration change, and the set of directories and their mount paths are unchanged.
### How was this patch tested?
A test added to `LocalDirsFeatureStepSuite` that runs `configurePod` ten times with four configured dirs and requires that not every run produces the same order. It also asserts, on each run, that the set of dirs is preserved, that the volume mounts are named `spark-local-dir-1..4`, and that the mount paths in order equal the `SPARK_LOCAL_DIRS` value, so a fix that scrambles the two out of step would fail.
Confirmed to fail against the unfixed tree, deterministically rather than flakily:
```
[info] - SPARK-XXXXX: randomize the local dirs resolved from configuration *** FAILED *** (29 milliseconds)
[info] 1 was not greater than 1 local dirs were never reordered across 10 runs: Set(/var/data/my-local-dir-1,/var/data/my-local-dir-2,/var/data/my-local-dir-3,/var/data/my-local-dir-4) (LocalDirsFeatureStepSuite.scala:135)
```
(That run predated the JIRA id, hence the placeholder in the test name.)
After the fix, a false failure would need all ten runs to draw the same permutation of four elements, about 1.4e-13.
`build/sbt -Pkubernetes 'kubernetes/testOnly org.apache.spark.deploy.k8s.features.LocalDirsFeatureStepSuite'`:
```
[info] - Resolve to default local dir if neither env nor configuration are set (52 milliseconds)
[info] - Use configured local dirs split on comma if provided. (3 milliseconds)
[info] - SPARK-58857: randomize the local dirs resolved from configuration (7 milliseconds)
[info] - Use tmpfs to back default local dir (1 millisecond)
[info] - local dir on mounted volume (6 milliseconds)
[info] Run completed in 901 milliseconds.
[info] Total number of tests run: 5
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 5, failed 0, canceled 0, ignored 0, pending 0
```
Because the fix makes both branches genuinely random, I ran the suite six times in total to check the pre-existing tests survive real reordering; all six runs were 5/5. `kubernetes/scalastyle` and `kubernetes/Test/scalastyle` report 0 errors.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Closes #58103 from LuciferYang/fix-k8s-localdirs-randomize.
Authored-by: YangJie <ya...@baidu.com>
Signed-off-by: yangjie01 <ya...@baidu.com>
(cherry picked from commit 350df6c1121d38c2d0a0c511690d8e433c6f6c4c)
Signed-off-by: yangjie01 <ya...@baidu.com>
---
.../deploy/k8s/features/LocalDirsFeatureStep.scala | 10 +++-----
.../k8s/features/LocalDirsFeatureStepSuite.scala | 30 ++++++++++++++++++++++
2 files changed, 34 insertions(+), 6 deletions(-)
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStep.scala
index b52ad732dfba..c80948343b23 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStep.scala
@@ -24,7 +24,6 @@ import io.fabric8.kubernetes.api.model._
import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod}
import org.apache.spark.deploy.k8s.Config._
-import org.apache.spark.util.ArrayImplicits._
import org.apache.spark.util.Utils.randomize
private[spark] class LocalDirsFeatureStep(
@@ -47,12 +46,11 @@ private[spark] class LocalDirsFeatureStep(
// exist in the image.
// We could make utils.getConfiguredLocalDirs opinionated about Kubernetes, as it is already
// a bit opinionated about YARN.
- val resolvedLocalDirs = Option(conf.sparkConf.getenv("SPARK_LOCAL_DIRS"))
+ val resolvedLocalDirs = randomize(Option(conf.sparkConf.getenv("SPARK_LOCAL_DIRS"))
.orElse(conf.getOption("spark.local.dir"))
.getOrElse(defaultLocalDir)
- .split(",")
- randomize(resolvedLocalDirs)
- localDirs = resolvedLocalDirs.toImmutableArraySeq
+ .split(","))
+ localDirs = resolvedLocalDirs
localDirVolumes = resolvedLocalDirs
.zipWithIndex
.map { case (_, index) =>
@@ -62,7 +60,7 @@ private[spark] class LocalDirsFeatureStep(
.withMedium(if (useLocalDirTmpFs) "Memory" else null)
.endEmptyDir()
.build()
- }.toImmutableArraySeq
+ }
localDirVolumeMounts = localDirVolumes
.zip(resolvedLocalDirs)
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala
index 3a9561051a89..452a2cb7922c 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/LocalDirsFeatureStepSuite.scala
@@ -16,6 +16,8 @@
*/
package org.apache.spark.deploy.k8s.features
+import scala.jdk.CollectionConverters._
+
import io.fabric8.kubernetes.api.model.{EnvVarBuilder, VolumeBuilder, VolumeMountBuilder}
import org.apache.spark.{SparkConf, SparkFunSuite}
@@ -105,6 +107,34 @@ class LocalDirsFeatureStepSuite extends SparkFunSuite {
.build())
}
+ test("SPARK-58857: randomize the local dirs resolved from configuration") {
+ // SPARK-39755 added randomization to both branches of configurePod, but the emptyDir branch
+ // called Utils.randomize as a statement and dropped its result, so the order stayed as
+ // configured. Run the step repeatedly and require that not every run agrees.
+ val dirs = (1 to 4).map(i => s"/var/data/my-local-dir-$i")
+ val sparkConf = new SparkConfWithEnv(Map("SPARK_LOCAL_DIRS" -> dirs.mkString(",")))
+ val kubernetesConf = KubernetesTestConf.createDriverConf(sparkConf = sparkConf)
+
+ val orders = (1 to 10).map { _ =>
+ val configuredPod =
+ new LocalDirsFeatureStep(kubernetesConf, defaultLocalDir).configurePod(
+ SparkPod.initialPod())
+ val env = configuredPod.container.getEnv.get(0)
+ assert(env.getName === "SPARK_LOCAL_DIRS")
+ // Whatever the order, the set of dirs is preserved and the mounts agree with the env var.
+ assert(env.getValue.split(",").sorted === dirs.sorted)
+ assert(configuredPod.pod.getSpec.getVolumes.size === dirs.size)
+ assert(configuredPod.container.getVolumeMounts.asScala.map(_.getName) ===
+ (1 to dirs.size).map(i => s"spark-local-dir-$i"))
+ assert(configuredPod.container.getVolumeMounts.asScala.map(_.getMountPath).mkString(",") ===
+ env.getValue)
+ env.getValue
+ }.toSet
+
+ // 10 runs of 4 dirs: a false failure needs the same permutation every time, (1/24)^9.
+ assert(orders.size > 1, s"local dirs were never reordered across 10 runs: $orders")
+ }
+
test("Use tmpfs to back default local dir") {
val sparkConf = new SparkConf(false).set(KUBERNETES_LOCAL_DIRS_TMPFS, true)
val kubernetesConf = KubernetesTestConf.createDriverConf(sparkConf = sparkConf)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org