[SPARK-37618][CORE][Followup] Support cleaning up shuffle blocks from external shuffle service

LICENSE-binary

@@ -546,6 +546,7 @@ jakarta.annotation:jakarta-annotation-api https://projects.eclipse.org/projects/
 jakarta.servlet:jakarta.servlet-api https://projects.eclipse.org/projects/ee4j.servlet
 jakarta.ws.rs:jakarta.ws.rs-api https://github.com/eclipse-ee4j/jaxrs-api
 org.glassfish.hk2.external:jakarta.inject
+com.github.jnr:jnr-posix
 
 
 Public Domain

core/pom.xml

@@ -61,6 +61,11 @@
       <groupId>com.twitter</groupId>
       <artifactId>chill-java</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.github.jnr</groupId>
+      <artifactId>jnr-posix</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.xbean</groupId>
       <artifactId>xbean-asm9-shaded</artifactId>

core/src/test/scala/org/apache/spark/storage/DiskBlockManagerSuite.scala

@@ -24,6 +24,7 @@ import java.util.HashMap
 
 import com.fasterxml.jackson.core.`type`.TypeReference
 import com.fasterxml.jackson.databind.ObjectMapper
+import jnr.posix.{POSIX, POSIXFactory}
 import org.apache.commons.io.FileUtils
 import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
 
@@ -141,28 +142,46 @@ class DiskBlockManagerSuite extends SparkFunSuite with BeforeAndAfterEach with B
     assert(attemptId.equals("1"))
   }
 
+  // Use jnr to get and override the current process umask.
+  // Expects the input mask to be an octal number
+  private def getAndSetUmask(posix: POSIX, mask: String): String = {
+    val prev = posix.umask(BigInt(mask, 8).toInt)
+    "0" + "%o".format(prev)
+  }
+
   test("SPARK-37618: Sub dirs are group writable when removing from shuffle service enabled") {
     val conf = testConf.clone
     conf.set("spark.local.dir", rootDirs)
     conf.set("spark.shuffle.service.enabled", "true")
     conf.set("spark.shuffle.service.removeShuffle", "false")
-    val diskBlockManager = new DiskBlockManager(conf, deleteFilesOnStop = true, isDriver = false)
-    val blockId = new TestBlockId("test")
-    val newFile = diskBlockManager.getFile(blockId)
-    val parentDir = newFile.getParentFile()
-    assert(parentDir.exists && parentDir.isDirectory)
-    val permission = Files.getPosixFilePermissions(parentDir.toPath)
-    assert(!permission.contains(PosixFilePermission.GROUP_WRITE))
-
-    assert(parentDir.delete())
-
-    conf.set("spark.shuffle.service.removeShuffle", "true")
-    val diskBlockManager2 = new DiskBlockManager(conf, deleteFilesOnStop = true, isDriver = false)
-    val newFile2 = diskBlockManager2.getFile(blockId)
-    val parentDir2 = newFile2.getParentFile()
-    assert(parentDir2.exists && parentDir2.isDirectory)
-    val permission2 = Files.getPosixFilePermissions(parentDir2.toPath)
-    assert(permission2.contains(PosixFilePermission.GROUP_WRITE))
+    val posix = POSIXFactory.getPOSIX
+
+    assume(posix.isNative, "Skipping test for SPARK-37618, native posix support not found")
+
+    val oldUmask = getAndSetUmask(posix, "077")
+    try {
+      val diskBlockManager = new DiskBlockManager(conf, deleteFilesOnStop = true,
+        isDriver = false)
+      val blockId = new TestBlockId("test")
+      val newFile = diskBlockManager.getFile(blockId)
+      val parentDir = newFile.getParentFile()
+      assert(parentDir.exists && parentDir.isDirectory)
+      val permission = Files.getPosixFilePermissions(parentDir.toPath)
+      assert(!permission.contains(PosixFilePermission.GROUP_WRITE))
+
+      assert(parentDir.delete())
+
+      conf.set("spark.shuffle.service.removeShuffle", "true")
+      val diskBlockManager2 = new DiskBlockManager(conf, deleteFilesOnStop = true,
+        isDriver = false)
+      val newFile2 = diskBlockManager2.getFile(blockId)
+      val parentDir2 = newFile2.getParentFile()
+      assert(parentDir2.exists && parentDir2.isDirectory)
+      val permission2 = Files.getPosixFilePermissions(parentDir2.toPath)
+      assert(permission2.contains(PosixFilePermission.GROUP_WRITE))
+    } finally {
+      getAndSetUmask(posix, oldUmask)
+    }
   }
 
   def writeToFile(file: File, numBytes: Int): Unit = {

pom.xml

@@ -457,6 +457,12 @@
         <artifactId>chill-java</artifactId>
         <version>${chill.version}</version>
       </dependency>
+      <dependency>
+        <groupId>com.github.jnr</groupId>
+        <artifactId>jnr-posix</artifactId>
+        <version>3.1.15</version>
+        <scope>test</scope>
+      </dependency>
       <!-- This artifact is a shaded version of ASM 9.x. The POM that was used to produce this
            is at https://github.com/apache/geronimo-xbean/tree/trunk/xbean-asm9-shaded
            For context on why we shade ASM, see SPARK-782 and SPARK-6152. -->