LuciferYang opened a new pull request, #57905:
URL: https://github.com/apache/spark/pull/57905
### 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 strict, one such entry also discards every correctly
nested directory configured alongside it. An empty string behaves the same way,
and the existing `filter(_ != null)` does not stop it because
`String.split(",")` yields `""` rather than null.
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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
