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 <[email protected]>
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 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
