This is an automated email from the ASF dual-hosted git repository.
LuciferYang 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 350df6c1121d [SPARK-58857][K8S] Bind the result of Utils.randomize in
LocalDirsFeatureStep
350df6c1121d is described below
commit 350df6c1121d38c2d0a0c511690d8e433c6f6c4c
Author: YangJie <[email protected]>
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 <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
