From 60211acccc60e35d94e04c4346c82042651f2826 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Tue, 1 Sep 2026 21:47:15 +0000 Subject: [PATCH 1/2] [CORE][MINOR] Scope standalone shuffle service local dirs to the registering app Nest executor local directories under //executor-* on the Worker, and add an opt-in config spark.shuffle.service.requireAppScopedLocalDirs (default false) that makes the external shuffle service reject executor registrations whose local directories do not lie inside the registering application's own per-application directory. The check is opt-in for rolling-upgrade compatibility: enable it only after every Worker is upgraded. The underlying LocalDirValidator already supported this mode (used by the YARN shuffle service); this wires standalone to use it. Co-authored-by: Cursor --- .../spark/deploy/ExternalShuffleService.scala | 13 +++-- .../apache/spark/deploy/worker/Worker.scala | 21 +++++++- .../spark/internal/config/package.scala | 16 ++++++ ...ExternalShuffleServiceLocalDirsSuite.scala | 50 +++++++++++++++++- .../spark/deploy/worker/WorkerSuite.scala | 52 +++++++++++++++++++ docs/spark-standalone.md | 17 ++++++ 6 files changed, 164 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala b/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala index 1f709ee758c0e..9d2d80185764e 100644 --- a/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala +++ b/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala @@ -83,8 +83,15 @@ class ExternalShuffleService(sparkConf: SparkConf, securityManager: SecurityMana /** Create a new shuffle block handler. Factored out for subclasses to override. */ protected def newShuffleBlockHandler(conf: TransportConf): ExternalBlockHandler = { - // Constrain registered localDirs to the configured local directories. + // Constrain registered localDirs to the configured local directories and, when + // spark.shuffle.service.requireAppScopedLocalDirs is enabled, to the registering + // application's own per-app directory (the Worker creates executor local dirs under a + // path containing the app id). The check is opt-in: executors launched by Workers that + // predate the per-app layout register unscoped paths and would be rejected, so enable it + // only once every Worker is upgraded. val localDirs = Utils.getConfiguredLocalDirs(sparkConf) + val requireAppScopedLocalDirs = + sparkConf.get(config.SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS) if (sparkConf.get(config.SHUFFLE_SERVICE_DB_ENABLED) && enabled) { val shuffleDBName = sparkConf.get(config.SHUFFLE_SERVICE_DB_BACKEND) logInfo( @@ -92,9 +99,9 @@ class ExternalShuffleService(sparkConf: SparkConf, securityManager: SecurityMana log"${MDC(SHUFFLE_DB_BACKEND_KEY, config.SHUFFLE_SERVICE_DB_BACKEND.key)}") new ExternalBlockHandler(conf, findRegisteredExecutorsDBFile(shuffleDBName.fileName(registeredExecutorsDB)), - localDirs, false) + localDirs, requireAppScopedLocalDirs) } else { - new ExternalBlockHandler(conf, null, localDirs, false) + new ExternalBlockHandler(conf, null, localDirs, requireAppScopedLocalDirs) } } diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala index 93fb64f485f62..f19b5df18b5e5 100755 --- a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala +++ b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala @@ -622,7 +622,17 @@ private[deploy] class Worker( val localRootDirs = Utils.getOrCreateLocalRootDirs(conf) val dirs = localRootDirs.flatMap { dir => try { - val appDir = Utils.createDirectory(dir, namePrefix = "executor") + // Nest executor local dirs under a per-application directory (the app id as + // a path segment) so the external shuffle service can require registered + // localDirs to be scoped to the registering application. + val appIdDir = new File(dir, appId) + appIdDir.mkdirs() + if (!appIdDir.isDirectory) { + throw new IOException(s"Failed to create directory $appIdDir") + } + Utils.chmod700(appIdDir) + val appDir = Utils.createDirectory(appIdDir.getAbsolutePath(), + namePrefix = "executor") Utils.chmod700(appDir) Some(appDir.getAbsolutePath()) } catch { @@ -777,6 +787,15 @@ private[deploy] class Worker( logInfo(log"Cleaning up local directories for application ${MDC(APP_ID, id)}") dirList.foreach { dir => Utils.deleteRecursively(new File(dir)) + // Executor dirs are nested under a per-application directory; remove it too + // once it is empty. + val appIdDir = new File(dir).getParentFile + if (appIdDir != null && appIdDir.getName == id) { + val remaining = appIdDir.list() + if (remaining != null && remaining.isEmpty) { + appIdDir.delete() + } + } } }(cleanupThreadExecutor).failed.foreach(e => logError(log"Clean up app dir ${MDC(PATHS, dirList)} failed", e) diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index 64a05fece570c..65fbf81a61cf5 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -908,6 +908,22 @@ package object config { .stringConf .createWithDefault("spark_shuffle") + private[spark] val SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS = + ConfigBuilder("spark.shuffle.service.requireAppScopedLocalDirs") + .doc("Whether the external shuffle service requires every local directory an executor " + + "reports at registration to lie inside the registering application's own " + + "per-application directory: the application id must appear as a path segment of the " + + "directory's canonical path, under one of the service's configured local directory " + + "roots. This keeps each application's shuffle and RDD blocks within its own directory " + + "scope at registration and cleanup time. Only affects standalone mode. Enable it only " + + "after every Worker in the cluster creates executor local directories under a " + + "per-application directory (Workers on this version do); executors launched by older " + + "Workers report unscoped paths and their registrations are rejected while this is " + + "enabled.") + .version("4.4.0") + .booleanConf + .createWithDefault(false) + private[spark] val KEYTAB = ConfigBuilder("spark.kerberos.keytab") .doc("Location of user's keytab.") .version("3.0.0") diff --git a/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceLocalDirsSuite.scala b/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceLocalDirsSuite.scala index 29576b5fd5ae6..c1ac9a089b40b 100644 --- a/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceLocalDirsSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceLocalDirsSuite.scala @@ -21,7 +21,10 @@ import java.io.File import java.util.UUID import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite} -import org.apache.spark.internal.config.{SHUFFLE_SERVICE_DB_ENABLED, SHUFFLE_SERVICE_ENABLED} +import org.apache.spark.internal.config.{ + SHUFFLE_SERVICE_DB_ENABLED, + SHUFFLE_SERVICE_ENABLED, + SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS} import org.apache.spark.util.Utils class ExternalShuffleServiceLocalDirsSuite extends SparkFunSuite { @@ -49,4 +52,49 @@ class ExternalShuffleServiceLocalDirsSuite extends SparkFunSuite { Utils.deleteRecursively(contained) } } + + test("requireAppScopedLocalDirs restricts localDirs to the registering app's directory") { + val sparkConf = new SparkConf() + .set(SHUFFLE_SERVICE_ENABLED, true) + .set(SHUFFLE_SERVICE_DB_ENABLED, false) + .set(SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS, true) + .set("spark.local.dir", System.getProperty("java.io.tmpdir")) + val service = new ExternalShuffleService(sparkConf, new SecurityManager(sparkConf)) + val handler = service.getBlockHandler + + val root = new File(Utils.getConfiguredLocalDirs(sparkConf).head) + val appId = s"app-${UUID.randomUUID()}" + val otherAppId = s"app-${UUID.randomUUID()}" + // The layout the patched Worker creates: //executor-*. + val appScoped = new File(new File(root, appId), "executor-1") + val otherAppScoped = new File(new File(root, otherAppId), "executor-1") + // The pre-upgrade Worker layout: directly under a root, no appId segment. + val unscoped = new File(root, s"blockmgr-${UUID.randomUUID()}") + assert(appScoped.mkdirs() && otherAppScoped.mkdirs() && unscoped.mkdirs()) + try { + // A localDir inside the registering application's own directory is accepted. + handler.validateLocalDirs(Array(appScoped.getAbsolutePath), appId) + + // A directory under a different application id is rejected even though it is contained + // under a configured root. + intercept[IllegalArgumentException] { + handler.validateLocalDirs(Array(otherAppScoped.getAbsolutePath), appId) + } + + // A contained directory without the appId as a path segment (what a Worker without + // the per-app layout would report) is rejected. + intercept[IllegalArgumentException] { + handler.validateLocalDirs(Array(unscoped.getAbsolutePath), appId) + } + + // Root containment is still enforced with app scoping on. + intercept[IllegalArgumentException] { + handler.validateLocalDirs(Array("/etc"), appId) + } + } finally { + Utils.deleteRecursively(new File(root, appId)) + Utils.deleteRecursively(new File(root, otherAppId)) + Utils.deleteRecursively(unscoped) + } + } } diff --git a/core/src/test/scala/org/apache/spark/deploy/worker/WorkerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/worker/WorkerSuite.scala index f9a0efce88708..9361e3d9cdfdf 100644 --- a/core/src/test/scala/org/apache/spark/deploy/worker/WorkerSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/worker/WorkerSuite.scala @@ -443,4 +443,56 @@ class WorkerSuite extends SparkFunSuite with Matchers with BeforeAndAfter with P assert(getHeartbeatTask(worker) == heartbeatTask) assert(getWorkDirCleanupTask(worker) == workDirCleanupTask) } + + test("app cleanup removes the per-application local-dir parent only when it becomes empty") { + val externalShuffleServiceSupplier = new Supplier[ExternalShuffleService] { + override def get: ExternalShuffleService = shuffleService + } + val worker = makeWorker(new SparkConf(), externalShuffleServiceSupplier) + val root = Utils.createTempDir(namePrefix = "worker-local-root") + try { + // New layout: LaunchExecutor creates executor local dirs as //executor-*. + // App A: its executor dir is the per-app parent's only entry, so cleanup must remove + // the parent as well. + val appA = "app-20260811000000-0001" + val appDirA = new File(root, appA) + val executorDirA = new File(appDirA, "executor-0001") + assert(executorDirA.mkdirs()) + + // App B: the per-app parent still holds another entry after the executor dir is + // deleted, so the parent must survive. + val appB = "app-20260811000000-0002" + val appDirB = new File(root, appB) + val executorDirB = new File(appDirB, "executor-0002") + assert(executorDirB.mkdirs()) + val leftoverB = new File(appDirB, "leftover") + assert(leftoverB.createNewFile()) + + // App C: pre-upgrade layout (no per-app parent). Cleanup must delete only the executor + // dir and never touch its parent, which is the shared local root. + val appC = "app-20260811000000-0003" + val executorDirC = new File(root, "executor-0003") + assert(executorDirC.mkdirs()) + + Seq((appA, executorDirA), (appB, executorDirB), (appC, executorDirC)).foreach { + case (appId, executorDir) => + worker.appDirectories(appId) = Seq(executorDir.getAbsolutePath) + worker.finishedApps += appId + worker.handleExecutorStateChanged( + ExecutorStateChanged(appId, 0, ExecutorState.EXITED, None, None)) + } + + eventually(timeout(1.second), interval(10.milliseconds)) { + assert(!executorDirA.exists()) + assert(!appDirA.exists()) + assert(!executorDirB.exists()) + assert(appDirB.exists()) + assert(leftoverB.exists()) + assert(!executorDirC.exists()) + assert(root.exists()) + } + } finally { + Utils.deleteRecursively(root) + } + } } diff --git a/docs/spark-standalone.md b/docs/spark-standalone.md index 149c79105a581..18cff8b53e2c1 100644 --- a/docs/spark-standalone.md +++ b/docs/spark-standalone.md @@ -501,6 +501,23 @@ SPARK_WORKER_OPTS supports the following system properties: 3.4.0 + + spark.shuffle.service.requireAppScopedLocalDirs + false + + When true, the external shuffle service only accepts executor registrations whose local + directories lie inside the registering application's own per-application directory (the + application id must appear as a path segment under one of the service's configured local + directory roots). This keeps each application's shuffle and RDD blocks within its own + directory scope at registration and cleanup time. Workers on this version always create + executor local directories as <root>/<appId>/executor-*, which satisfies the + check; that layout is additive and harmless while the check is disabled. For a rolling + upgrade, first upgrade every Worker, then set this to true on the shuffle service: enabling + it while executors launched by older Workers are still registering rejects their + registrations. This only affects standalone mode. + + 4.4.0 + spark.storage.cleanupFilesAfterExecutorExit true From ebf927dd9cd4cca2a6d6a08aa4f08a7398399bcf Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Tue, 8 Sep 2026 06:00:41 -0700 Subject: [PATCH 2/2] Clarify --- .../scala/org/apache/spark/deploy/ExternalShuffleService.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala b/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala index 9d2d80185764e..f44ca7b167b4b 100644 --- a/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala +++ b/core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala @@ -88,7 +88,7 @@ class ExternalShuffleService(sparkConf: SparkConf, securityManager: SecurityMana // application's own per-app directory (the Worker creates executor local dirs under a // path containing the app id). The check is opt-in: executors launched by Workers that // predate the per-app layout register unscoped paths and would be rejected, so enable it - // only once every Worker is upgraded. + // only once every Spark application is upgraded. val localDirs = Utils.getConfiguredLocalDirs(sparkConf) val requireAppScopedLocalDirs = sparkConf.get(config.SHUFFLE_SERVICE_REQUIRE_APP_SCOPED_LOCAL_DIRS)