SPARK-49297: DO NOT MERGE!

.github/workflows/build_main.yml

@@ -30,3 +30,5 @@ jobs:
       packages: write
     name: Run
     uses: ./.github/workflows/build_and_test.yml
+    with:
+      java: 21

common/utils/src/main/java/org/apache/spark/network/util/JavaUtils.java

@@ -158,6 +158,7 @@ private static void deleteRecursivelyUsingUnixNative(File file) throws IOExcepti
     ProcessBuilder builder = new ProcessBuilder("rm", "-rf", file.getAbsolutePath());
     Process process = null;
     int exitCode = -1;
+    logger.error("Attila: deleteRecursivelyUsingUnixNative: " + file.getName());
 
     try {
       // In order to avoid deadlocks, consume the stdout (and stderr) of the process

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -1044,6 +1044,7 @@ private[spark] object Utils
    * Throws an exception if deletion is unsuccessful.
    */
   override def deleteRecursively(file: File): Unit = {
+    logInfo(s"Attila deleteRecursively: $file")
     super.deleteRecursively(file)
     if (file != null) {
       ShutdownHookManager.removeShutdownDeleteDir(file)

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

@@ -18,6 +18,8 @@
 package org.apache.spark.storage
 
 import java.io.File
+import java.nio.file.Files
+import java.nio.file.Paths
 import java.util.concurrent.{ConcurrentHashMap, ConcurrentLinkedQueue, Semaphore, TimeUnit}
 
 import scala.collection.mutable.ArrayBuffer
@@ -361,10 +363,12 @@ class BlockManagerDecommissionIntegrationSuite extends SparkFunSuite with LocalS
     val sparkTempDir = System.getProperty("java.io.tmpdir")
 
     def shuffleFiles: Seq[File] = {
-      FileUtils
+      val r = FileUtils
         .listFiles(new File(sparkTempDir), Array("data", "index"), true)
         .asScala
         .toSeq
+      logInfo(s"Attila: $r")
+      r
     }
 
     val existingShuffleFiles = shuffleFiles
@@ -377,20 +381,22 @@ class BlockManagerDecommissionIntegrationSuite extends SparkFunSuite with LocalS
       .set(config.STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED, true)
     sc = new SparkContext(conf)
     TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
-    val shuffleBlockUpdates = new ArrayBuffer[BlockId]()
-    var isDecommissionedExecutorRemoved = false
+    val shuffleBlockUpdates = new ConcurrentLinkedQueue[BlockId]()
     val execToDecommission = sc.getExecutorIds().head
+    val decommissionedExecutorLocalDir = sc.parallelize(1 to 10, 100).flatMap {  _ =>
+      if (SparkEnv.get.executorId == execToDecommission) {
+        SparkEnv.get.blockManager.getLocalDiskDirs
+      } else {
+        Array.empty[String]
+      }
+    }.collect().toSet
+    assert(decommissionedExecutorLocalDir.size == 1)
     sc.addSparkListener(new SparkListener {
       override def onBlockUpdated(blockUpdated: SparkListenerBlockUpdated): Unit = {
         if (blockUpdated.blockUpdatedInfo.blockId.isShuffle) {
-          shuffleBlockUpdates += blockUpdated.blockUpdatedInfo.blockId
+          shuffleBlockUpdates.add(blockUpdated.blockUpdatedInfo.blockId)
         }
       }
-
-      override def onExecutorRemoved(executorRemoved: SparkListenerExecutorRemoved): Unit = {
-        assert(execToDecommission === executorRemoved.executorId)
-        isDecommissionedExecutorRemoved = true
-      }
     })
 
     // Run a job to create shuffle data
@@ -409,12 +415,13 @@ class BlockManagerDecommissionIntegrationSuite extends SparkFunSuite with LocalS
       )
 
     eventually(timeout(1.minute), interval(10.milliseconds)) {
-      assert(isDecommissionedExecutorRemoved)
+      assert(Files.notExists(Paths.get(decommissionedExecutorLocalDir.head)))
       // Ensure there are shuffle data have been migrated
       assert(shuffleBlockUpdates.size >= 2)
     }
 
     val shuffleId = shuffleBlockUpdates
+      .asScala
       .find(_.isInstanceOf[ShuffleIndexBlockId])
       .map(_.asInstanceOf[ShuffleIndexBlockId].shuffleId)
       .get

dev/run-tests.py

@@ -680,4 +680,4 @@ def _test():
 
 if __name__ == "__main__":
     _test()
-    main()
+    exec_sbt(["project core; testOnly org.apache.spark.storage.BlockManagerDecommissionIntegrationSuite"])