You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/08/28 22:19:48 UTC
(spark) branch branch-4.1 updated: [SPARK-58750][4.1][CORE] Tolerate FileAlreadyExistsException from checkpoint part file rename
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new f30d7a326999 [SPARK-58750][4.1][CORE] Tolerate FileAlreadyExistsException from checkpoint part file rename
f30d7a326999 is described below
commit f30d7a3269996a44bd1c9b1b61169ed247e73dd8
Author: James Willis <ja...@wherobots.com>
AuthorDate: Fri Aug 28 15:19:37 2026 -0700
[SPARK-58750][4.1][CORE] Tolerate FileAlreadyExistsException from checkpoint part file rename
Backport of #57976 to branch-4.1.
### What changes were proposed in this pull request?
`ReliableCheckpointRDD.writePartitionToCheckpointFile` renames the attempt-temp file onto the final part file and only handles a rename that reports failure by returning `false` (HDFS semantics). This PR additionally catches `FileAlreadyExistsException` from that rename and routes it into the same existing handling: if the final part file exists, another attempt of this task already committed it, so the temp file is deleted and the write is treated as successful. If the destination do [...]
### Why are the changes needed?
Since [HADOOP-16721](https://issues.apache.org/jira/browse/HADOOP-16721) (Hadoop 3.3.1), S3A deliberately raises `FileAlreadyExistsException` when the rename destination is an existing file, instead of returning `false` as HDFS does. ABFS behaves the same way. The Hadoop FileSystem specification does not guarantee HDFS-style `false` reporting.
Under speculative execution (or a zombie attempt racing a retry), two attempts of the same checkpoint task race to rename onto the same final part file. On HDFS the loser sees `rename() == false` and Spark correctly treats it as "some other copy of this task must've finished before us". On S3A/ABFS the loser gets an unhandled `FileAlreadyExistsException`, which fails the task — and because the destination now permanently exists, **every retry of that task fails on the same rename**, s [...]
Observed in production (Spark 4.0.1, Hadoop 3.4.1, `spark.speculation=true`):
```
org.apache.hadoop.fs.FileAlreadyExistsException: Failed to rename
s3://<bucket>/<prefix>/spark-checkpoints/<uuid>/rdd-207/.part-00379-attempt-25364
to s3://<bucket>/<prefix>/spark-checkpoints/<uuid>/rdd-207/part-00379; destination file exists
at org.apache.hadoop.fs.s3a.S3AFileSystem.initiateRename(S3AFileSystem.java:2468)
at org.apache.hadoop.fs.s3a.S3AFileSystem.rename(S3AFileSystem.java:2392)
at org.apache.spark.rdd.ReliableCheckpointRDD$.writePartitionToCheckpointFile(ReliableCheckpointRDD.scala:229)
...
ERROR TaskSetManager: Task 379 in stage 105.0 failed 4 times; aborting job
```
See [SPARK-58750](https://issues.apache.org/jira/browse/SPARK-58750) for full details. Structured Streaming's `CheckpointFileManager` was already hardened for divergent rename semantics; the RDD checkpoint writer is the remaining caller assuming HDFS semantics.
### Does this PR introduce _any_ user-facing change?
No. RDD checkpointing to S3A/ABFS under speculative execution (or task retry after a committed rename) now succeeds instead of unrecoverably failing the job, which is the bug fix itself.
### How was this patch tested?
Added a regression test to `CheckpointStorageSuite` that writes the same checkpoint partition from two task attempts against a `FileSystem` mimicking S3A's rename semantics (raises `FileAlreadyExistsException` when the destination file exists). Without the fix, the second attempt throws and would fail the task; with the fix, it succeeds and exactly one committed part file remains, with the attempt-temp file cleaned up.
Ran `build/sbt "core/testOnly org.apache.spark.CheckpointStorageSuite"` locally.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (model claude-fable-5)
Closes #58358 from james-willis/backport-SPARK-58750-4.1.
Authored-by: James Willis <ja...@wherobots.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../apache/spark/rdd/ReliableCheckpointRDD.scala | 16 ++++++-
.../scala/org/apache/spark/CheckpointSuite.scala | 51 +++++++++++++++++++++-
2 files changed, 64 insertions(+), 3 deletions(-)
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 fd42cea795d6..ee5cb63e14bb 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
@@ -225,7 +225,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 58512a2282ac..d894639c949b 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.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)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org