This is an automated email from the ASF dual-hosted git repository.
viirya 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 363e944af0c0 [SPARK-60603][CORE] Evict shuffle service canonical local
dir cache on application removal
363e944af0c0 is described below
commit 363e944af0c036d125446ee53cef1fc47910a137
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Wed Oct 28 10:17:17 2026 -0700
[SPARK-60603][CORE] Evict shuffle service canonical local dir cache on
application removal
### What changes were proposed in this pull request?
This is a follow-up to 18f88fba9d0 ("[CORE][MINOR] Improve ESS file
consistency"), which added read-time block path verification to
`ExternalShuffleBlockResolver` (`spark.shuffle.service.verifyBlockPathsAtRead`,
default `true`).
- `applicationRemoved()` now evicts the removed executors' local dirs from
`canonicalLocalDirs`, keeping any dir that is still referenced by another
registered executor. `executorRemoved()` does not evict, matching how
`executors` is handled there: the shuffle service keeps serving a removed
executor's shuffle files until its application is removed.
- The index file path is now verified only on a `shuffleIndexCache` miss,
i.e. when the index file is actually read. The data file is still verified on
every fetch, since it is opened on every fetch.
- Document `spark.shuffle.service.verifyBlockPathsAtRead` in
`docs/configuration.md`.
### Why are the changes needed?
`canonicalLocalDirs` is filled in `registerExecutor()` and lazily in
`getVerifiedFile()`, but nothing ever removes entries. On YARN every executor's
local dirs are unique (`.../appcache/<appId>/blockmgr-<uuid>`), so a
long-running NodeManager-hosted shuffle service grows the map with every
executor it has ever served.
Also, each shuffle block fetch did two `getCanonicalFile()` calls (index
and data) before the index cache lookup, so the cache did not save the index
path canonicalization even when the index file was not read.
### Does this PR introduce _any_ user-facing change?
No behavior change other than fixing the unbounded memory growth in the
shuffle service. The config was previously undocumented and is now documented.
### How was this patch tested?
New tests in `ExternalShuffleBlockResolverSuite`:
- `canonicalLocalDirsEvictedOnApplicationRemoved`: registers 2 apps x 100
executors plus a dir shared by both apps, checks that `executorRemoved` keeps
entries, that removing one app keeps the shared dir, and that removing both
empties the map. It fails without this fix (201 entries left instead of 101).
- `testIndexFileResolvingOutsideLocalDirsRejected`: an index file that is a
symlink outside the registered local dirs is still rejected when it is loaded
into the index cache.
Ran `ExternalShuffleBlockResolverSuite`, `ExternalBlockHandlerSuite`,
`ExternalShuffleCleanupSuite` and `ExternalShuffleIntegrationSuite` on JDK 17.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
Closes #59851 from viirya/SPARK-60603.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../shuffle/ExternalShuffleBlockResolver.java | 43 ++++++++++++--
.../shuffle/ExternalShuffleBlockResolverSuite.java | 69 ++++++++++++++++++++++
docs/configuration.md | 11 ++++
3 files changed, 118 insertions(+), 5 deletions(-)
diff --git
a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java
b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java
index 0e7ddaa12204..b7a6cf50550f 100644
---
a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java
+++
b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java
@@ -95,8 +95,10 @@ public class ExternalShuffleBlockResolver {
private final boolean verifyBlockPathsAtRead;
// Canonical forms of registered executor local dirs, cached because they
are checked on
- // every block open. Keyed by the raw registered path string.
- private final ConcurrentMap<String, Path> canonicalLocalDirs = new
ConcurrentHashMap<>();
+ // every block open. Keyed by the raw registered path string. Entries are
evicted when the
+ // last registered executor referencing the dir is removed in
applicationRemoved().
+ @VisibleForTesting
+ final ConcurrentMap<String, Path> canonicalLocalDirs = new
ConcurrentHashMap<>();
@VisibleForTesting
final File registeredExecutorFile;
@@ -243,6 +245,7 @@ public class ExternalShuffleBlockResolver {
logger.info("Application {} removed, cleanupLocalDirs = {}",
MDC.of(LogKeys.APP_ID, appId),
MDC.of(LogKeys.CLEANUP_LOCAL_DIRS, cleanupLocalDirs));
+ Set<String> removedLocalDirs = new HashSet<>();
Iterator<Map.Entry<AppExecId, ExecutorShuffleInfo>> it =
executors.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<AppExecId, ExecutorShuffleInfo> entry = it.next();
@@ -252,6 +255,9 @@ public class ExternalShuffleBlockResolver {
// Only touch executors associated with the appId that was removed.
if (appId.equals(fullId.appId)) {
it.remove();
+ if (executor.localDirs != null) {
+ removedLocalDirs.addAll(Arrays.asList(executor.localDirs));
+ }
if (db != null &&
AppsWithRecoveryDisabled.isRecoveryEnabledForApp(fullId.appId)) {
try {
db.delete(dbAppExecKey(fullId));
@@ -271,6 +277,25 @@ public class ExternalShuffleBlockResolver {
}
}
}
+ evictCanonicalLocalDirs(removedLocalDirs);
+ }
+
+ /**
+ * Drops the cached canonical forms of the given local dirs, except those
still referenced by
+ * a registered executor. Without this, the cache would grow with every
application served.
+ */
+ private void evictCanonicalLocalDirs(Set<String> localDirs) {
+ if (localDirs.isEmpty() || canonicalLocalDirs.isEmpty()) {
+ return;
+ }
+ for (ExecutorShuffleInfo executor : executors.values()) {
+ if (executor.localDirs != null) {
+ for (String localDir : executor.localDirs) {
+ localDirs.remove(localDir);
+ }
+ }
+ }
+ canonicalLocalDirs.keySet().removeAll(localDirs);
}
/**
@@ -378,11 +403,19 @@ public class ExternalShuffleBlockResolver {
*/
private ManagedBuffer getSortBasedShuffleBlockData(
ExecutorShuffleInfo executor, int shuffleId, long mapId, int
startReduceId, int endReduceId) {
- String indexFilePath = getVerifiedFile(
- executor, "shuffle_" + shuffleId + "_" + mapId + "_0.index").getPath();
+ String indexFileName = "shuffle_" + shuffleId + "_" + mapId + "_0.index";
+ String indexFilePath = ExecutorDiskUtils.getFilePath(
+ executor.localDirs, executor.subDirsPerLocalDir, indexFileName);
try {
- ShuffleIndexInformation shuffleIndexInformation =
shuffleIndexCache.get(indexFilePath);
+ // The index file is only read when it is loaded into the cache, so
verify its path only
+ // on a cache miss. The data file is opened on every fetch and is
verified every time.
+ ShuffleIndexInformation shuffleIndexInformation =
+ shuffleIndexCache.getIfPresent(indexFilePath);
+ if (shuffleIndexInformation == null) {
+ getVerifiedFile(executor, indexFileName);
+ shuffleIndexInformation = shuffleIndexCache.get(indexFilePath);
+ }
ShuffleIndexRecord shuffleIndexRecord = shuffleIndexInformation.getIndex(
startReduceId, endReduceId);
return new FileSegmentManagedBuffer(
diff --git
a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolverSuite.java
b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolverSuite.java
index b2cb4d63ebdc..ef81d68cbb99 100644
---
a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolverSuite.java
+++
b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolverSuite.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
+import java.nio.file.StandardCopyOption;
import java.util.UUID;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -140,6 +141,74 @@ public class ExternalShuffleBlockResolverSuite {
}
}
+ @Test
+ public void testIndexFileResolvingOutsideLocalDirsRejected() throws
IOException {
+ // The index file is verified when it is loaded into the index cache.
+ File outsideFile = File.createTempFile("outside", ".index");
+ TestShuffleDataContext context = new TestShuffleDataContext(2, 5);
+ context.create();
+ try {
+ context.insertSortShuffleData(1, 0, new byte[][] {
+ "AAAA".getBytes(StandardCharsets.UTF_8),
+ "BBBB".getBytes(StandardCharsets.UTF_8)});
+ ExternalShuffleBlockResolver resolver = new
ExternalShuffleBlockResolver(conf, null);
+ resolver.registerExecutor("app1", "exec0",
context.createExecutorInfo(SORT_MANAGER));
+
+ // Replace the index file with a link to a valid copy outside the
registered local dirs.
+ File indexFile = new File(ExecutorDiskUtils.getFilePath(
+ context.localDirs, context.subDirsPerLocalDir, "shuffle_1_0_0.index"));
+ Files.copy(indexFile.toPath(), outsideFile.toPath(),
+ StandardCopyOption.REPLACE_EXISTING);
+ assertTrue(indexFile.delete());
+ try {
+ Files.createSymbolicLink(indexFile.toPath(), outsideFile.toPath());
+ } catch (UnsupportedOperationException | IOException e) {
+ // Filesystems without symlink support cannot exercise this case.
+ Assumptions.assumeTrue(false, "symlinks unsupported on this
filesystem");
+ }
+
+ RuntimeException e = assertThrows(RuntimeException.class,
+ () -> resolver.getBlockData("app1", "exec0", 1, 0, 0));
+ assertTrue(e.getMessage().contains("resolves outside"), "Bad error
message: " + e);
+ } finally {
+ context.cleanup();
+ outsideFile.delete();
+ }
+ }
+
+ @Test
+ public void canonicalLocalDirsEvictedOnApplicationRemoved() throws
IOException {
+ ExternalShuffleBlockResolver resolver =
+ new ExternalShuffleBlockResolver(conf, null, Runnable::run);
+ String root = new File(System.getProperty("java.io.tmpdir"),
UUID.randomUUID().toString())
+ .getAbsolutePath();
+ String sharedDir = root + "/shared";
+ int numExecutors = 100;
+ for (String appId : new String[] {"app1", "app2"}) {
+ for (int i = 0; i < numExecutors; i++) {
+ String[] localDirs = {root + "/" + appId + "/exec" + i, sharedDir};
+ resolver.registerExecutor(appId, "exec" + i,
+ new ExecutorShuffleInfo(localDirs, 64, SORT_MANAGER));
+ }
+ }
+ assertEquals(2 * numExecutors + 1, resolver.canonicalLocalDirs.size());
+
+ // Executor removal keeps the executor registered, so its dirs stay cached.
+ for (int i = 0; i < numExecutors; i++) {
+ resolver.executorRemoved("exec" + i, "app1");
+ }
+ assertEquals(2 * numExecutors + 1, resolver.canonicalLocalDirs.size());
+
+ // A dir still referenced by another application's executor is kept.
+ resolver.applicationRemoved("app1", false);
+ assertEquals(numExecutors + 1, resolver.canonicalLocalDirs.size());
+ assertTrue(resolver.canonicalLocalDirs.containsKey(sharedDir));
+
+ resolver.applicationRemoved("app2", false);
+ assertTrue(resolver.canonicalLocalDirs.isEmpty());
+ assertEquals(0, resolver.getRegisteredExecutorsSize());
+ }
+
@Test
public void jsonSerializationOfExecutorRegistration() throws IOException {
ObjectMapper mapper = new ObjectMapper();
diff --git a/docs/configuration.md b/docs/configuration.md
index 5643586fba24..3ff96ad8c5d0 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -1561,6 +1561,17 @@ Apart from these, the following properties are also
available, and may be useful
</td>
<td>3.0.0</td>
</tr>
+<tr>
+ <td><code>spark.shuffle.service.verifyBlockPathsAtRead</code></td>
+ <td>true</td>
+ <td>
+ Whether the ExternalShuffleService re-resolves the path of a shuffle or
RDD block file when
+ serving it, and refuses the file if the resolved path is no longer under
the executor's
+ registered local directories. Disabling it skips the path canonicalization
done on each
+ block fetch.
+ </td>
+ <td>4.1.4</td>
+</tr>
<tr>
<td><code>spark.shuffle.service.db.enabled</code></td>
<td>true</td>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
