[SPARK-59332][CORE] Scope standalone shuffle service local dirs

core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala

@@ -83,18 +83,25 @@ 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 Spark application 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(
         log"Use ${MDC(SHUFFLE_DB_BACKEND_NAME, shuffleDBName.name())} as the implementation of " +
         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)
     }
   }
 

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)

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")

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: <root>/<appId>/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)
+    }
+  }
 }

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 <root>/<appId>/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)
+    }
+  }
 }

docs/spark-standalone.md

@@ -501,6 +501,23 @@ SPARK_WORKER_OPTS supports the following system properties:
   </td>
   <td>3.4.0</td>
 </tr>
+<tr>
+  <td><code>spark.shuffle.service.requireAppScopedLocalDirs</code></td>
+  <td>false</td>
+  <td>
+    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 <code>&lt;root&gt;/&lt;appId&gt;/executor-*</code>, 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.
+  </td>
+  <td>4.4.0</td>
+</tr>
 <tr>
   <td><code>spark.storage.cleanupFilesAfterExecutorExit</code></td>
   <td>true</td>