diff --git a/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala b/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala index 156a94abd6281..244d79aa217a4 100644 --- a/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala +++ b/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 @@ -226,7 +226,19 @@ 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(log"Rename from ${MDC(TEMP_OUTPUT_PATH, tempOutputPath)} to" + + log" ${MDC(FINAL_OUTPUT_PATH, finalOutputPath)} failed", e) + false + } + if (!renamed) { if (!fs.exists(finalOutputPath)) { logInfo(log"Deleting tempOutputPath ${MDC(TEMP_OUTPUT_PATH, tempOutputPath)}") fs.delete(tempOutputPath, false) diff --git a/core/src/test/scala/org/apache/spark/CheckpointSuite.scala b/core/src/test/scala/org/apache/spark/CheckpointSuite.scala index 976e9b94d23cd..af06e8f29f3b7 100644 --- a/core/src/test/scala/org/apache/spark/CheckpointSuite.scala +++ b/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.{LocalFileSystem, Path} +import org.apache.hadoop.fs.{FileAlreadyExistsException, LocalFileSystem, 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() @@ -752,3 +784,20 @@ class MkdirsFailingFilesystem extends LocalFileSystem { if (f.getName.startsWith("rdd-")) false else super.mkdirs(f) } } + +/** + * 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) + } +}