[SPARK-52615][CORE] Replace File.mkdirs with Utils.createDirectory

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

We hit an issue that `File.mkdirs()` may occasionally fail with no error during the submission phase(we didn't configure `spark.yarn.archive` in that cluster so each submission requires packaging and uploading spark client jars, which cosumes a lot of disk IO), which was also mentioned in `Utils.createDirectory`

> // SPARK-35907: The check was required by File.mkdirs() because it could sporadically
> // fail silently. ...

```
25/06/27 19:12:17 WARN Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.
25/06/27 19:12:17 ERROR Utils: Failed to create dir in ./local. Ignoring this directory.
25/06/27 19:12:17 INFO Client: Deleted staging directory hdfs://<cluster>/user/<user>/.sparkStaging/application_1747844918192_28291290
Exception in thread "main" java.io.IOException: Failed to get a temp directory under [./local].
	at org.apache.spark.util.Utils$.getLocalDir(Utils.scala:896)
	at org.apache.spark.deploy.yarn.Client.prepareLocalResources(Client.scala:672)
	at org.apache.spark.deploy.yarn.Client.createContainerLaunchContext(Client.scala:1005)
	at org.apache.spark.deploy.yarn.Client.submitApplication(Client.scala:231)
	at org.apache.spark.deploy.yarn.Client.run(Client.scala:1352)
	at org.apache.spark.deploy.yarn.YarnClusterApplication.start(Client.scala:1800)
	at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:1019)
	at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:192)
	at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:215)
	at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:91)
	at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1107)
	at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1116)
	at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
```

So I replaced `File.mkdirs` with `Utils.createDirectory` and deployed it into our internal cluster, no similar failures happens after then ... (not sure why, maybe the replaced NIO method is more robust?)

Additional context:
[JDK-4227544](https://bugs.openjdk.org/browse/JDK-4227544) "design bug: File.mkdir(), etc. don't provide reason for failure" get closed with "Won't Fix"

> It is too late to change this now. The new io framework will handle this situation better.

### Why are the changes needed?

To achieve better error message reporting when creating a directory fails.

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

No.

### How was this patch tested?

I made such a change in internal Spark, and deployed it to a busy YARN cluster, the submit process has been stable so far.

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

No.

Closes #51322 from pan3793/SPARK-52615.

Authored-by: Cheng Pan <chengpan@apache.org>
Signed-off-by: yangjie01 <yangjie01@baidu.com>

Author: pan3793

common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/TestShuffleDataContext.java

@@ -22,6 +22,7 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.nio.file.Files;
 
 import com.google.common.io.Closeables;
 
@@ -54,7 +55,7 @@ public void create() throws IOException {
       localDirs[i] = JavaUtils.createDirectory(root, "spark").getAbsolutePath();
 
       for (int p = 0; p < subDirsPerLocalDir; p ++) {
-        new File(localDirs[i], String.format("%02x", p)).mkdirs();
+        Files.createDirectories(new File(localDirs[i], String.format("%02x", p)).toPath());
       }
     }
   }

common/utils/src/test/scala/org/apache/spark/util/IvyTestUtils.scala

@@ -95,7 +95,7 @@ private[spark] object IvyTestUtils {
       className: String,
       packageName: String): Seq[(String, File)] = {
     val rFilesDir = new File(dir, "R" + File.separator + "pkg")
-    new File(rFilesDir, "R").mkdirs()
+    SparkFileUtils.createDirectory(new File(rFilesDir, "R"))
     val contents =
       s"""myfunc <- function(x) {
         |  SparkR:::callJStatic("$packageName.$className", "myFunc", x)
@@ -150,11 +150,11 @@ private[spark] object IvyTestUtils {
       useIvyLayout: Boolean): File = {
     if (useIvyLayout) {
       val ivyXmlPath = pathFromCoordinate(artifact, tempPath, "ivy", true)
-      ivyXmlPath.mkdirs()
+      SparkFileUtils.createDirectory(ivyXmlPath)
       createIvyDescriptor(ivyXmlPath, artifact, dependencies)
     } else {
       val pomPath = pathFromCoordinate(artifact, tempPath, "pom", useIvyLayout)
-      pomPath.mkdirs()
+      SparkFileUtils.createDirectory(pomPath)
       createPom(pomPath, artifact, dependencies)
     }
   }
@@ -293,13 +293,13 @@ private[spark] object IvyTestUtils {
     // Where the root of the repository exists, and what Ivy will search in
     val tempPath = tempDir.getOrElse(SparkFileUtils.createTempDir())
     // Create directory if it doesn't exist
-    tempPath.mkdirs()
+    SparkFileUtils.createDirectory(tempPath)
     // Where to create temporary class files and such
     val root = new File(tempPath, tempPath.hashCode().toString)
-    root.mkdirs()
+    SparkFileUtils.createDirectory(root)
     try {
       val jarPath = pathFromCoordinate(artifact, tempPath, "jar", useIvyLayout)
-      jarPath.mkdirs()
+      SparkFileUtils.createDirectory(jarPath)
       val className = "MyLib"
 
       val javaClass = createJavaClass(root, className, artifact.groupId)

connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala

@@ -1899,7 +1899,7 @@ abstract class AvroSuite
     withTempPath { tempDir =>
       val tempEmptyDir = s"$tempDir/sqlOverwrite"
       // Create a temp directory for table that will be overwritten
-      new File(tempEmptyDir).mkdirs()
+      Utils.createDirectory(tempEmptyDir)
       spark.sql(
         s"""
            |CREATE TEMPORARY VIEW episodes

connector/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaRDDSuite.scala

@@ -38,6 +38,7 @@ import org.scalatest.concurrent.Eventually.{eventually, interval, timeout}
 import org.apache.spark._
 import org.apache.spark.scheduler.ExecutorCacheTaskLocation
 import org.apache.spark.streaming.kafka010.mocks.MockTime
+import org.apache.spark.util.Utils
 
 class KafkaRDDSuite extends SparkFunSuite {
 
@@ -91,7 +92,7 @@ class KafkaRDDSuite extends SparkFunSuite {
     val logs = new Pool[TopicPartition, UnifiedLog]()
     val logDir = kafkaTestUtils.brokerLogDir
     val dir = new File(logDir, topic + "-" + partition)
-    dir.mkdirs()
+    Utils.createDirectory(dir)
     val logProps = new ju.Properties()
     logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT)
     logProps.put(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG, java.lang.Float.valueOf(0.1f))

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

@@ -153,7 +153,7 @@ private[deploy] object RPackageUtils extends Logging {
           if (verbose) {
             print(log"Creating directory: ${MDC(PATH, dir)}", printStream)
           }
-          dir.mkdirs
+          Utils.createDirectory(dir)
         } else {
           val inStream = jar.getInputStream(entry)
           val outPath = new File(tempDir, entryPath)

core/src/main/scala/org/apache/spark/deploy/worker/DriverRunner.scala

@@ -145,7 +145,7 @@ private[deploy] class DriverRunner(
    */
   private def createWorkingDirectory(): File = {
     val driverDir = new File(workDir, driverId)
-    if (!driverDir.exists() && !driverDir.mkdirs()) {
+    if (!driverDir.exists() && !Utils.createDirectory(driverDir)) {
       throw new IOException("Failed to create directory " + driverDir)
     }
     driverDir

core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala

@@ -600,7 +600,7 @@ private[deploy] class Worker(
 
           // Create the executor's working directory
           val executorDir = new File(workDir, appId + "/" + execId)
-          if (!executorDir.mkdirs()) {
+          if (!Utils.createDirectory(executorDir)) {
             throw new IOException("Failed to create directory " + executorDir)
           }
 

core/src/main/scala/org/apache/spark/rdd/PipedRDD.scala

@@ -90,7 +90,7 @@ private[spark] class PipedRDD[T: ClassTag](
       val currentDir = new File(".")
       logDebug("currentDir = " + currentDir.getAbsolutePath())
       val taskDirFile = new File(taskDirectory)
-      taskDirFile.mkdirs()
+      Utils.createDirectory(taskDirFile)
 
       try {
         val tasksDirFilter = new NotEqualsFileNameFilter("tasks")

core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala

@@ -316,7 +316,7 @@ private[spark] class DiskBlockManager(
         throw SparkCoreErrors.failToCreateDirectoryError(dirToCreate.getAbsolutePath, maxAttempts)
       }
       try {
-        dirToCreate.mkdirs()
+        Utils.createDirectory(dirToCreate)
         Files.setPosixFilePermissions(
           dirToCreate.toPath, PosixFilePermissions.fromString("rwxrwx---"))
         if (dirToCreate.exists()) {

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

@@ -458,7 +458,7 @@ private[spark] object Utils
    * to work around a security issue, see also SPARK-38631.
    */
   private def unTarUsingJava(source: File, dest: File): Unit = {
-    if (!dest.mkdirs && !dest.isDirectory) {
+    if (!Utils.createDirectory(dest) && !dest.isDirectory) {
       throw new IOException(s"Mkdirs failed to create $dest")
     } else {
       try {
@@ -810,7 +810,7 @@ private[spark] object Utils
     configuredLocalDirs.flatMap { root =>
       try {
         val rootDir = new File(root)
-        if (rootDir.exists || rootDir.mkdirs()) {
+        if (rootDir.exists || Utils.createDirectory(rootDir)) {
           val dir = createTempDir(root)
           chmod700(dir)
           Some(dir.getAbsolutePath)