[SPARK-53037][CORE][SQL][K8S] Support closeQuietly in SparkErrorUtils

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

This PR aims to support `closeQuietly` in `SparkErrorUtils`.

### Why are the changes needed?

To provide a way to close a closable object without any exceptions and logs.

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

No behavior change.

### How was this patch tested?

Pass the CIs.

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

No.

Closes #51742 from dongjoon-hyun/SPARK-53037.

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

Author: dongjoon-hyun

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

@@ -105,6 +105,17 @@ private[spark] trait SparkErrorUtils extends Logging {
     }
     new String(out.toByteArray, UTF_8)
   }
+
+  /** Try to close by ignoring all exceptions. This is different from JavaUtils.closeQuietly. */
+  def closeQuietly(closeable: Closeable): Unit = {
+    if (closeable != null) {
+      try {
+        closeable.close()
+      } catch {
+        case _: Exception =>
+      }
+    }
+  }
 }
 
 private[spark] object SparkErrorUtils extends SparkErrorUtils

core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java

@@ -27,7 +27,6 @@
 import java.util.function.Supplier;
 
 import com.google.common.annotations.VisibleForTesting;
-import org.apache.commons.io.IOUtils;
 
 import org.apache.spark.TaskContext;
 import org.apache.spark.executor.ShuffleWriteMetrics;
@@ -886,7 +885,7 @@ public void close() throws IOException {
 
     private void closeIfPossible(UnsafeSorterIterator iterator) {
       if (iterator instanceof Closeable closeable) {
-        IOUtils.closeQuietly((closeable));
+        Utils.closeQuietly((closeable));
       }
     }
   }

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

@@ -36,7 +36,6 @@ import scala.util.control.NonFatal
 import com.codahale.metrics.{MetricRegistry, MetricSet}
 import com.esotericsoftware.kryo.KryoException
 import com.google.common.cache.CacheBuilder
-import org.apache.commons.io.IOUtils
 
 import org.apache.spark._
 import org.apache.spark.errors.SparkCoreErrors
@@ -373,7 +372,7 @@ private[spark] class BlockManager(
           logInfo(extendMessageWithBlockDetails(ex.getMessage, blockId))
           throw ex
       } finally {
-        IOUtils.closeQuietly(inputStream)
+        Utils.closeQuietly(inputStream)
       }
     }
 

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

@@ -30,7 +30,6 @@ import scala.collection.mutable.{ArrayBuffer, HashMap, HashSet, Queue}
 import scala.util.{Failure, Success}
 
 import io.netty.util.internal.OutOfDirectMemoryError
-import org.apache.commons.io.IOUtils
 import org.roaringbitmap.RoaringBitmap
 
 import org.apache.spark.{MapOutputTracker, SparkException, TaskContext}
@@ -1408,7 +1407,7 @@ private class BufferReleasingInputStream(
         val diagnosisResponse = checkedInOpt.map { checkedIn =>
           iterator.diagnoseCorruption(checkedIn, address, blockId)
         }
-        IOUtils.closeQuietly(this)
+        Utils.closeQuietly(this)
         // We'd never retry the block whatever the cause is since the block has been
         // partially consumed by downstream RDDs.
         iterator.throwFetchFailedException(blockId, mapIndex, address, e, diagnosisResponse)

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

@@ -3084,8 +3084,8 @@ private[spark] object Utils
       logDebug(log"Unzipped from ${MDC(PATH, dfsZipFile)}\n\t${MDC(PATHS, files.mkString("\n\t"))}")
     } finally {
       // Close everything no matter what happened
-      IOUtils.closeQuietly(in)
-      IOUtils.closeQuietly(out)
+      Utils.closeQuietly(in)
+      Utils.closeQuietly(out)
     }
     files.toSeq
   }

core/src/main/scala/org/apache/spark/util/logging/RollingFileAppender.scala

@@ -27,6 +27,7 @@ import org.apache.spark.SparkConf
 import org.apache.spark.internal.{config, MDC}
 import org.apache.spark.internal.LogKeys._
 import org.apache.spark.util.ArrayImplicits._
+import org.apache.spark.util.Utils
 
 /**
  * Continuously appends data from input stream into the given file, and rolls
@@ -96,8 +97,8 @@ private[spark] class RollingFileAppender(
         gzOutputStream.close()
         activeFile.delete()
       } finally {
-        IOUtils.closeQuietly(inputStream)
-        IOUtils.closeQuietly(gzOutputStream)
+        Utils.closeQuietly(inputStream)
+        Utils.closeQuietly(gzOutputStream)
       }
     } else {
       Files.move(activeFile, rolloverFile)

core/src/test/scala/org/apache/spark/JobArtifactSetSuite.scala

@@ -22,6 +22,7 @@ import java.util.zip.{ZipEntry, ZipOutputStream}
 
 import org.apache.commons.io.IOUtils
 
+import org.apache.spark.util.Utils
 
 class JobArtifactSetSuite extends SparkFunSuite with LocalSparkContext {
 
@@ -33,8 +34,8 @@ class JobArtifactSetSuite extends SparkFunSuite with LocalSparkContext {
     val zipEntry = new ZipEntry(fileToZip.getName)
     zipOut.putNextEntry(zipEntry)
     IOUtils.copy(fis, zipOut)
-    IOUtils.closeQuietly(fis)
-    IOUtils.closeQuietly(zipOut)
+    Utils.closeQuietly(fis)
+    Utils.closeQuietly(zipOut)
   }
 
   test("JobArtifactSet uses resources from SparkContext") {

core/src/test/scala/org/apache/spark/util/FileAppenderSuite.scala

@@ -35,6 +35,7 @@ import org.scalatest.BeforeAndAfter
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
 import org.apache.spark.internal.config
+import org.apache.spark.util.Utils
 import org.apache.spark.util.logging.{FileAppender, RollingFileAppender, SizeBasedRollingPolicy, TimeBasedRollingPolicy}
 
 class FileAppenderSuite extends SparkFunSuite with BeforeAndAfter {
@@ -389,7 +390,7 @@ class FileAppenderSuite extends SparkFunSuite with BeforeAndAfter {
         try {
           IOUtils.toString(inputStream, StandardCharsets.UTF_8)
         } finally {
-          IOUtils.closeQuietly(inputStream)
+          Utils.closeQuietly(inputStream)
         }
       } else {
         Files.asCharSource(file, StandardCharsets.UTF_8).read()

core/src/test/scala/org/apache/spark/util/UtilsSuite.scala

@@ -248,8 +248,8 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties {
         assert(mergedStream.read() === -1)
         assert(byteBufferInputStream.chunkedByteBuffer === null)
       } finally {
-        IOUtils.closeQuietly(mergedStream)
-        IOUtils.closeQuietly(in)
+        Utils.closeQuietly(mergedStream)
+        Utils.closeQuietly(in)
       }
     }
   }

resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/Utils.scala

@@ -33,6 +33,7 @@ import org.apache.commons.io.output.ByteArrayOutputStream
 
 import org.apache.spark.{SPARK_VERSION, SparkException}
 import org.apache.spark.internal.Logging
+import org.apache.spark.util.SparkErrorUtils
 
 object Utils extends Logging {
 
@@ -146,8 +147,8 @@ object Utils extends Logging {
     val zipEntry = new ZipEntry(fileToZip.getName)
     zipOut.putNextEntry(zipEntry)
     IOUtils.copy(fis, zipOut)
-    IOUtils.closeQuietly(fis)
-    IOUtils.closeQuietly(zipOut)
+    SparkErrorUtils.closeQuietly(fis)
+    SparkErrorUtils.closeQuietly(zipOut)
   }
 
   def createTarGzFile(inFile: String, outFile: String): Unit = {