[SPARK-53137][CORE][SQL] Support forceDeleteOnExit in SparkFileUtils and JavaUtils

### What changes were proposed in this pull request?

This PR aims to support `forceDeleteOnExit` in `SparkFileUtils` and `JavaUtils`.

### Why are the changes needed?

To improve Spark's file utility functions.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Pass the CIs.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #51866 from dongjoon-hyun/SPARK-53137.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dongjoon-hyun

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

@@ -78,6 +78,31 @@ public static void deleteQuietly(File file) {
     }
   }
 
+  /** Registers the file or directory for deletion when the JVM exists. */
+  public static void forceDeleteOnExit(File file) throws IOException {
+    if (file != null && file.exists()) {
+      if (!file.isDirectory()) {
+        file.deleteOnExit();
+      } else {
+        Path path = file.toPath();
+        Files.walkFileTree(path, new SimpleFileVisitor<Path>() {
+          @Override
+          public FileVisitResult preVisitDirectory(Path p, BasicFileAttributes a)
+              throws IOException {
+            p.toFile().deleteOnExit();
+            return a.isSymbolicLink() ? FileVisitResult.SKIP_SUBTREE : FileVisitResult.CONTINUE;
+          }
+
+          @Override
+          public FileVisitResult visitFile(Path p, BasicFileAttributes a) throws IOException {
+            p.toFile().deleteOnExit();
+            return FileVisitResult.CONTINUE;
+          }
+        });
+      }
+    }
+  }
+
   /** Move a file from src to dst. */
   public static void moveFile(File src, File dst) throws IOException {
     if (src == null || dst == null || !src.exists() || src.isDirectory() || dst.exists()) {

common/utils/src/main/scala/org/apache/spark/util/SparkFileUtils.scala

@@ -153,6 +153,11 @@ private[spark] trait SparkFileUtils extends Logging {
     JavaUtils.deleteQuietly(file)
   }
 
+  /** Registers the file or directory for deletion when the JVM exists. */
+  def forceDeleteOnExit(file: File): Unit = {
+    JavaUtils.forceDeleteOnExit(file)
+  }
+
   def getFile(names: String*): File = {
     require(names != null && names.forall(_ != null))
     names.tail.foldLeft(Path.of(names.head)) { (path, part) =>

dev/checkstyle.xml

@@ -215,6 +215,10 @@
             <property name="format" value="FileUtils\.forceDelete\("/>
             <property name="message" value="Use deleteRecursively of JavaUtils/SparkFileUtils/Utils instead." />
         </module>
+        <module name="RegexpSinglelineJava">
+            <property name="format" value="FileUtils\.forceDeleteOnExit\("/>
+            <property name="message" value="Use forceDeleteOnExit of JavaUtils/SparkFileUtils/Utils instead." />
+        </module>
         <module name="RegexpSinglelineJava">
             <property name="format" value="FileUtils\.deleteQuietly"/>
             <property name="message" value="Use deleteQuietly of JavaUtils/SparkFileUtils/Utils instead." />

scalastyle-config.xml

@@ -322,6 +322,11 @@ This file is divided into 3 sections:
     <customMessage>Use deleteRecursively of JavaUtils/SparkFileUtils/Utils</customMessage>
   </check>
 
+  <check customId="forceDeleteOnExit" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
+    <parameters><parameter name="regex">\bFileUtils\.forceDeleteOnExit\b</parameter></parameters>
+    <customMessage>Use forceDeleteOnExit of JavaUtils/SparkFileUtils/Utils instead.</customMessage>
+  </check>
+
   <check customId="deleteQuietly" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
     <parameters><parameter name="regex">\bFileUtils\.deleteQuietly\b</parameter></parameters>
     <customMessage>Use deleteQuietly of JavaUtils/SparkFileUtils/Utils</customMessage>

sql/hive-thriftserver/src/main/java/org/apache/hive/service/cli/session/SessionManager.java

@@ -28,7 +28,6 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
-import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
 import org.apache.hive.service.CompositeService;
@@ -43,6 +42,7 @@
 import org.apache.spark.internal.SparkLoggerFactory;
 import org.apache.spark.internal.LogKeys;
 import org.apache.spark.internal.MDC;
+import org.apache.spark.network.util.JavaUtils;
 import org.apache.spark.util.Utils;
 
 /**
@@ -138,7 +138,7 @@ private void initOperationLogRootDir() {
       LOG.info("Operation log root directory is created: {}",
         MDC.of(LogKeys.PATH, operationLogRootDir.getAbsolutePath()));
       try {
-        FileUtils.forceDeleteOnExit(operationLogRootDir);
+        JavaUtils.forceDeleteOnExit(operationLogRootDir);
       } catch (IOException e) {
         LOG.warn("Failed to schedule cleanup HS2 operation logging root dir: {}", e,
           MDC.of(LogKeys.PATH, operationLogRootDir.getAbsolutePath()));