[SPARK-58750][CORE] Tolerate FileAlreadyExistsException from checkpoint part file rename

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

@@ -24,7 +24,7 @@ import scala.reflect.ClassTag
 import scala.util.control.NonFatal
 
 import com.google.common.cache.{CacheBuilder, CacheLoader}
-import org.apache.hadoop.fs.Path
+import org.apache.hadoop.fs.{FileAlreadyExistsException, Path}
 
 import org.apache.spark._
 import org.apache.spark.broadcast.Broadcast
@@ -225,7 +225,18 @@ private[spark] object ReliableCheckpointRDD extends Logging {
       serializeStream.close()
     })
 
-    if (!fs.rename(tempOutputPath, finalOutputPath)) {
+    // On HDFS, renaming onto an existing destination reports failure by returning false, which
+    // is handled below. Some FileSystem implementations instead raise FileAlreadyExistsException
+    // (e.g. S3A since HADOOP-16721, ABFS); treat it the same way, as it means another attempt of
+    // this task has already committed the final output (SPARK-58750).
+    val renamed = try {
+      fs.rename(tempOutputPath, finalOutputPath)
+    } catch {
+      case e: FileAlreadyExistsException =>
+        logDebug(s"Rename from $tempOutputPath to $finalOutputPath failed", e)
+        false
+    }
+    if (!renamed) {
       if (!fs.exists(finalOutputPath)) {
         logInfo(log"Deleting tempOutputPath ${MDC(TEMP_OUTPUT_PATH, tempOutputPath)}")
         fs.delete(tempOutputPath, false)

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

@@ -18,18 +18,22 @@
 package org.apache.spark
 
 import java.io.File
+import java.net.URI
+import java.util.Properties
 
 import scala.reflect.ClassTag
 
-import org.apache.hadoop.fs.Path
+import org.apache.hadoop.fs.{FileAlreadyExistsException, Path, RawLocalFileSystem}
 
 import org.apache.spark.internal.config.CACHE_CHECKPOINT_PREFERRED_LOCS_EXPIRE_TIME
 import org.apache.spark.internal.config.UI._
 import org.apache.spark.io.CompressionCodec
+import org.apache.spark.memory.TaskMemoryManager
 import org.apache.spark.rdd._
 import org.apache.spark.shuffle.FetchFailedException
 import org.apache.spark.storage.{BlockId, StorageLevel, TestBlockId}
 import org.apache.spark.util.ArrayImplicits._
+import org.apache.spark.util.SerializableConfiguration
 import org.apache.spark.util.Utils
 
 trait RDDCheckpointTester { self: SparkFunSuite =>
@@ -669,6 +673,34 @@ class CheckpointStorageSuite extends SparkFunSuite with LocalSparkContext {
     }
   }
 
+  test("SPARK-58750: checkpointing tolerates FileAlreadyExistsException on part file rename") {
+    withTempDir { checkpointDir =>
+      val conf = new SparkConf().set(UI_ENABLED.key, "false")
+      sc = new SparkContext("local", "test", conf)
+      sc.hadoopConfiguration.set(
+        "fs.faee.impl", classOf[FileAlreadyExistsRenameFileSystem].getName)
+      val broadcastedConf = SerializableConfiguration.broadcast(sc, sc.hadoopConfiguration)
+      val outputDir = s"faee://${checkpointDir.getAbsolutePath}"
+
+      def writePartition(taskAttemptId: Long, attemptNumber: Int): Unit = {
+        val ctx = new TaskContextImpl(0, 0, 0, taskAttemptId, attemptNumber, 1,
+          new TaskMemoryManager(sc.env.memoryManager, 0L), new Properties, sc.env.metricsSystem)
+        ReliableCheckpointRDD.writePartitionToCheckpointFile[Int](
+          outputDir, broadcastedConf)(ctx, Iterator(1, 2, 3))
+      }
+
+      writePartition(taskAttemptId = 0L, attemptNumber = 0)
+      // A speculative or retried attempt of the same partition finds the part file already
+      // committed by the first attempt. On filesystems that raise FileAlreadyExistsException
+      // from rename (S3A, ABFS), this must be treated as success rather than fail the task.
+      writePartition(taskAttemptId = 1L, attemptNumber = 1)
+
+      val fs = new Path(outputDir).getFileSystem(sc.hadoopConfiguration)
+      val fileNames = fs.listStatus(new Path(outputDir)).map(_.getPath.getName)
+      assert(fileNames === Array("part-00000"))
+    }
+  }
+
   test("SPARK-48268: checkpoint directory via configuration") {
     withTempDir { checkpointDir =>
       val conf = new SparkConf()
@@ -685,3 +717,20 @@ class CheckpointStorageSuite extends SparkFunSuite with LocalSparkContext {
     }
   }
 }
+
+/**
+ * A local filesystem mimicking how some Hadoop FileSystem implementations report a rename onto
+ * an existing file: by raising FileAlreadyExistsException (e.g. S3A since HADOOP-16721, ABFS)
+ * rather than returning false as HDFS does.
+ */
+class FileAlreadyExistsRenameFileSystem extends RawLocalFileSystem {
+  override def getUri: URI = URI.create("faee:///")
+
+  override def rename(src: Path, dst: Path): Boolean = {
+    if (exists(dst)) {
+      throw new FileAlreadyExistsException(
+        s"Failed to rename $src to $dst; destination file exists")
+    }
+    super.rename(src, dst)
+  }
+}