This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 09d12485d677 [SPARK-58693][K8S] Skip unlistable directories in K8s
shuffle data recovery
09d12485d677 is described below
commit 09d12485d6772b2d37561a672c0c598ec42057e6
Author: YangJie <[email protected]>
AuthorDate: Tue Aug 11 16:11:43 2026 +0800
[SPARK-58693][K8S] Skip unlistable directories in K8s shuffle data recovery
### What changes were proposed in this pull request?
Make the local-disk shuffle recovery scan in
`KubernetesLocalDiskShuffleExecutorComponents.recoverDiskStore` tolerate
directories it cannot list, instead of aborting the whole recovery.
All five listings now go through a helper that logs and skips when
`File.listFiles` returns null. A local directory too shallow to have a volume
root two levels up is skipped with a warning that names the config to change.
The dead `if (files != null)` guard on an `Array.flatMap` result is removed.
`SPARK-57530` took the same approach for `SparkFileUtils.recursiveList`.
### Why are the changes needed?
The scan is a strict `Array.map`/`flatMap` chain with no null handling on
any of its five `File.listFiles` calls, and the two-level `getParent` walk is
unguarded. One bad entry throws, `initializeExecutor` swallows the exception
through `Utils.tryLogNonFatalError`, and the executor goes on to serve requests
with nothing recovered. The only trace is an NPE stack trace that does not
mention shuffle recovery, and the job recomputes every map output on the reused
PVC.
Two cases reach it.
A configured local directory with fewer than three path components. The
scan walks two levels up from each entry, so for `/data` the first `getParent`
yields `/` and the second yields null, and `new File((null: String))` throws
`NullPointerException` from `java.io.File.<init>`. `/data` is the mount path in
the code block immediately above the recovery instructions in the Local Storage
section of `running-on-kubernetes.md`, so this is reachable by following the
docs. Because `map` is s [...]
A directory in the walk whose `listFiles()` returns null: removed while the
walk runs, or present but not readable by the executor uid.
The only null check in the method, `if (files != null)` applied to the
result of `Array.flatMap`, can never fire, so the walk reads as null-safe while
none of the five listings are guarded.
`SPARK-40459` already established that a single file should not block the
rest of the recovery. The collection phase in front of that loop was still
all-or-nothing.
### Does this PR introduce _any_ user-facing change?
No API or configuration change. Recovery that previously aborted on either
condition now proceeds and recovers the remaining files, and each skip logs a
warning.
`spark.shuffle.sort.io.plugin.class=org.apache.spark.shuffle.KubernetesLocalDiskShuffleDataIO`
plus `spark.kubernetes.driver.reusePersistentVolumeClaim=true` are still
required for any of this code to run.
### How was this patch tested?
Two tests added to `KubernetesLocalDiskShuffleDataIOSuite`, one per
condition, both driving `recoverDiskStore` directly.
Each test stubs `BlockManager.TempFileBasedBlockStoreUpdater` to throw a
sentinel `IllegalStateException` and asserts that message, so passing requires
the scan to have walked down to a recoverable shuffle file. A test that merely
completed without throwing would pass on an empty scan and prove nothing.
Both were confirmed to fail against the unfixed tree before the fix was
written, with the NPE arriving from the expected place in each case:
```
[info] - SPARK-58693: a local dir with fewer than three path components is
skipped *** FAILED *** (571 milliseconds)
[info] Expected exception java.lang.IllegalStateException to be thrown,
but java.lang.NullPointerException was thrown
(KubernetesLocalDiskShuffleDataIOSuite.scala:280)
[info] Cause: java.lang.NullPointerException:
[info] at java.base/java.io.File.<init>(File.java:278)
[info] - SPARK-58693: an unlistable directory in the scan does not abort
recovery *** FAILED *** (8 milliseconds)
[info] Expected exception java.lang.IllegalStateException to be thrown,
but java.lang.NullPointerException was thrown
(KubernetesLocalDiskShuffleDataIOSuite.scala:298)
[info] Cause: java.lang.NullPointerException: Cannot invoke
"scala.collection.IterableOnce.knownSize()" because "xs" is null
[info] at
scala.collection.mutable.ArrayBuilder.addAll(ArrayBuilder.scala:69)
[info] at
scala.collection.mutable.ArrayBuilder.addAll(ArrayBuilder.scala:24)
```
The unlistable-directory test uses `setReadable(false, false)` and,
following `FsHistoryProviderSuite`, `assume`s the bit actually cleared so it
self-skips for root. The scala `kubernetes` module runs in the `build` job,
which has no `container:` and therefore does not run as root, so it does
execute in CI. The other test needs no permission manipulation and runs
everywhere.
`build/sbt -Pkubernetes 'kubernetes/testOnly
org.apache.spark.shuffle.KubernetesLocalDiskShuffleDataIOSuite'`:
```
[info] - recompute is not blocked by the recovery (12 seconds, 384
milliseconds)
[info] - Partial recompute shuffle data (11 seconds, 118 milliseconds)
[info] - A new rdd and full recovery of old data (8 seconds, 816
milliseconds)
[info] - Multi stages (8 seconds, 557 milliseconds)
[info] - SPARK-44501: Ignore checksum files (571 milliseconds)
[info] - SPARK-44534: Handle only shuffle files (7 milliseconds)
[info] - SPARK-58693: a local dir with fewer than three path components is
skipped (6 milliseconds)
[info] - SPARK-58693: an unlistable directory in the scan does not abort
recovery (4 milliseconds)
[info] Run completed in 42 seconds, 491 milliseconds.
[info] Total number of tests run: 8
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 8, failed 0, canceled 0, ignored 0, pending 0
```
`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 #57905 from LuciferYang/fix-k8s-shuffle-recovery-listfiles.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
(cherry picked from commit 595a88249d6cc584ed2d2a728b9b2b94034d870f)
Signed-off-by: yangjie01 <[email protected]>
---
...ernetesLocalDiskShuffleExecutorComponents.scala | 54 ++++++++++++++++++----
.../KubernetesLocalDiskShuffleDataIOSuite.scala | 52 +++++++++++++++++++++
2 files changed, 97 insertions(+), 9 deletions(-)
diff --git
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/shuffle/KubernetesLocalDiskShuffleExecutorComponents.scala
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/shuffle/KubernetesLocalDiskShuffleExecutorComponents.scala
index cbe215c3f218..3d3b10bcecba 100644
---
a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/shuffle/KubernetesLocalDiskShuffleExecutorComponents.scala
+++
b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/shuffle/KubernetesLocalDiskShuffleExecutorComponents.scala
@@ -71,6 +71,42 @@ class
KubernetesLocalDiskShuffleExecutorComponents(sparkConf: SparkConf)
}
object KubernetesLocalDiskShuffleExecutorComponents extends Logging {
+ /**
+ * `File.listFiles` returns null when the directory cannot be read (an IO
error, a permission
+ * denial, or the directory being removed during the walk). Skip such a
directory with a warning
+ * instead of failing the whole recovery, so one unreadable entry cannot
cost us every other
+ * recoverable file. This is the single-level counterpart of
`SparkFileUtils.recursiveList`, and
+ * unlike `JavaUtils.listFilesSafely` it does not throw on an unlistable
directory.
+ */
+ private def listEntriesOrEmpty(dir: File): Array[File] = {
+ val entries = dir.listFiles()
+ if (entries != null) {
+ entries
+ } else {
+ logWarning(log"Failed to list ${MDC(LogKeys.FILE_ABSOLUTE_PATH,
dir.getAbsolutePath)}; " +
+ log"skipping it during shuffle data recovery.")
+ Array.empty[File]
+ }
+ }
+
+ /**
+ * Walks two levels up from a local directory to reach the volume root
shared by all executor
+ * generations. Returns None for a path with fewer than three components,
where `getParent`
+ * yields null.
+ */
+ private def volumeRootOf(localDir: String): Option[File] = {
+ val root = Option(new File(localDir).getParentFile).flatMap(p =>
Option(p.getParentFile))
+ if (root.isEmpty) {
+ logWarning(log"Cannot locate the volume root of local directory " +
+ log"${MDC(LogKeys.PATH, localDir)}; skipping it during shuffle data
recovery. Recovery " +
+ log"requires a local directory nested at least two levels below the
volume root, such as " +
+ log"/data/spark-x/executor-y. Set the mount path of the
spark-local-dir-* volume, or " +
+ log"${MDC(LogKeys.CONFIG, "spark.local.dir")} when no such volume is
mounted, " +
+ log"accordingly.")
+ }
+ root
+ }
+
/**
* This tries to recover shuffle data of dead executors' local dirs if
exists.
* Since the executors are already dead, we cannot use `getHostLocalDirs`.
@@ -79,18 +115,18 @@ object KubernetesLocalDiskShuffleExecutorComponents
extends Logging {
def recoverDiskStore(conf: SparkConf, bm: BlockManager): Unit = {
// Find All files
val (checksumFiles, files) = Utils.getConfiguredLocalDirs(conf)
- .filter(_ != null)
- .map(s => new File(new File(new File(s).getParent).getParent))
+ .filter(s => s != null && s.nonEmpty)
+ .flatMap(volumeRootOf)
.flatMap { dir =>
- val oldDirs = dir.listFiles().filter { f =>
+ val oldDirs = listEntriesOrEmpty(dir).filter { f =>
f.isDirectory && f.getName.startsWith("spark-")
}
- val files = oldDirs
- .flatMap(_.listFiles).filter(_.isDirectory) // executor-xxx
- .flatMap(_.listFiles).filter(_.isDirectory) // blockmgr-xxx
- .flatMap(_.listFiles).filter(_.isDirectory) // 00
- .flatMap(_.listFiles)
- if (files != null) files.toImmutableArraySeq else Seq.empty
+ oldDirs
+ .flatMap(listEntriesOrEmpty).filter(_.isDirectory) // executor-xxx
+ .flatMap(listEntriesOrEmpty).filter(_.isDirectory) // blockmgr-xxx
+ .flatMap(listEntriesOrEmpty).filter(_.isDirectory) // 00
+ .flatMap(listEntriesOrEmpty)
+ .toImmutableArraySeq
}
.partition(_.getName.contains(".checksum"))
val (indexFiles, dataFiles) = files.partition(_.getName.endsWith(".index"))
diff --git
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/shuffle/KubernetesLocalDiskShuffleDataIOSuite.scala
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/shuffle/KubernetesLocalDiskShuffleDataIOSuite.scala
index 25b2cad9ddd2..ba1777719afb 100644
---
a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/shuffle/KubernetesLocalDiskShuffleDataIOSuite.scala
+++
b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/shuffle/KubernetesLocalDiskShuffleDataIOSuite.scala
@@ -252,4 +252,56 @@ class KubernetesLocalDiskShuffleDataIOSuite extends
SparkFunSuite with LocalRoot
when(bm.TempFileBasedBlockStoreUpdater).thenAnswer(_ => throw new
Exception())
KubernetesLocalDiskShuffleExecutorComponents.recoverDiskStore(sparkConf,
bm)
}
+
+ // A sentinel thrown from the block-store updater. Reaching it proves the
scan actually
+ // walked down to a recoverable shuffle file, so the test cannot pass
vacuously.
+ private val sentinelMessage = "reached a shuffle file"
+
+ private def sentinelBlockManager(): BlockManager = {
+ val bm = mock(classOf[BlockManager])
+ when(bm.TempFileBasedBlockStoreUpdater)
+ .thenAnswer(_ => throw new IllegalStateException(sentinelMessage))
+ bm
+ }
+
+ private def createRecoverableShuffleFile(localDir: String): Unit = {
+ val dir = new File(localDir, "blockmgr-z/00")
+ Files.createDirectories(dir.toPath())
+ Files.write(new File(dir, "shuffle_0_0_0.index").toPath(), Array[Byte](0,
0, 0, 0))
+ }
+
+ test("SPARK-58693: a local dir with fewer than three path components is
skipped") {
+ val deepDir = conf.get("spark.local.dir") + "/spark-x/executor-y"
+ createRecoverableShuffleFile(deepDir)
+ // "/data" is what the Local Storage example in running-on-kubernetes.md
configures, and
+ // walking two levels up from it yields null. The deep dir must still be
recovered.
+ val sparkConf = conf.clone.set("spark.local.dir", s"/data,$deepDir")
+
+ val m = intercept[IllegalStateException] {
+ KubernetesLocalDiskShuffleExecutorComponents
+ .recoverDiskStore(sparkConf, sentinelBlockManager())
+ }.getMessage
+ assert(m.contains(sentinelMessage))
+ }
+
+ test("SPARK-58693: an unlistable directory in the scan does not abort
recovery") {
+ val deepDir = conf.get("spark.local.dir") + "/spark-x/executor-y"
+ createRecoverableShuffleFile(deepDir)
+ // Shaped like the ext4 lost+found that sits at the root of a freshly
formatted PVC: a
+ // directory the executor can see but not list, so listFiles() returns
null.
+ val unlistable = new File(conf.get("spark.local.dir") +
"/spark-x/lost+found")
+ Files.createDirectories(unlistable.toPath())
+ assert(unlistable.setReadable(false, false))
+ assume(!unlistable.canRead, "requires an unreadable directory; skipped
when run as root")
+
+ try {
+ val m = intercept[IllegalStateException] {
+ KubernetesLocalDiskShuffleExecutorComponents
+ .recoverDiskStore(conf.clone.set("spark.local.dir", deepDir),
sentinelBlockManager())
+ }.getMessage
+ assert(m.contains(sentinelMessage))
+ } finally {
+ unlistable.setReadable(true, false)
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
