You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Dongjoon Hyun (Jira)" <ji...@apache.org> on 2026/08/27 18:03:00 UTC
[jira] [Updated] (SPARK-58750) Tolerate FileAlreadyExistsException from checkpoint part file rename
[ https://issues.apache.org/jira/browse/SPARK-58750?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Dongjoon Hyun updated SPARK-58750:
----------------------------------
Summary: Tolerate FileAlreadyExistsException from checkpoint part file rename (was: RDD checkpoint to S3A fails unrecoverably under speculation/task retry)
> Tolerate FileAlreadyExistsException from checkpoint part file rename
> --------------------------------------------------------------------
>
> Key: SPARK-58750
> URL: https://issues.apache.org/jira/browse/SPARK-58750
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
> Affects Versions: 3.2.0, 3.2.1, 3.3.0, 3.2.2, 3.3.1, 3.2.3, 3.2.4, 3.3.3, 3.4.2, 3.3.2, 3.4.0, 3.4.1, 3.5.0, 3.5.1, 3.3.4, 3.5.2, 3.4.3, 3.4.4, 3.5.3, 3.5.4, 3.5.5, 4.1.0, 3.5.6, 4.0.0, 4.0.1, 3.5.7, 4.0.2, 3.5.8, 4.2.0, 4.1.1, 4.1.2, 3.5.9, 4.0.3, 4.2.1, 3.5.10
> Environment: I produced this error in Spark 4.0.1 on Kubernetes, Hadoop 3.4.1 (S3A), with {{spark.speculation=true}}
> Reporter: James Willis
> Assignee: James Willis
> Priority: Major
> Labels: pull-request-available
> Fix For: 4.4.0
>
>
> h2. Summary
> {{ReliableCheckpointRDD.writePartitionToCheckpointFile}} assumes HDFS rename semantics: it handles {{fs.rename()}} returning {{false}} when the destination exists, but does not handle the rename {*}throwing{*}. Since HADOOP-16721 (Hadoop 3.3.1), S3A deliberately raises {{FileAlreadyExistsException}} when the rename destination is an existing file, instead of returning {{{}false{}}}. ABFS behaves the same way.
> Under speculative execution, two attempts of the same checkpoint task race to rename their attempt-temp file onto the same final part file. On HDFS the loser gets {{rename() == false}} and Spark correctly treats it as "some other copy of this task must've finished before us". On S3A the loser gets an unhandled {{{}FileAlreadyExistsException{}}}, which fails the task even though the data is already written. Because the destination file now permanently exists, {*}every subsequent retry of that task fails on the same rename{*}, so {{spark.task.maxFailures}} is always exhausted and the job aborts.
> Every Spark release since the default Hadoop dependency moved to 3.3.1+ (Spark 3.2.0) is affected when checkpointing to S3A or ABFS with speculation enabled. The code is unchanged in master.
> h2. Observed failure (production, Spark 4.0.1 / Hadoop 3.4.1)
> {noformat}
> org.apache.hadoop.fs.FileAlreadyExistsException: Failed to rename
> s3://<bucket>/<prefix>/spark-checkpoints/44151dc0-c680-446d-8556-f6aef47d73dd/rdd-207/.part-00379-attempt-25364
> to s3://<bucket>/<prefix>/spark-checkpoints/44151dc0-c680-446d-8556-f6aef47d73dd/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.innerRename(S3AFileSystem.java:2533)
> at org.apache.hadoop.fs.s3a.S3AFileSystem.lambda$rename$6(S3AFileSystem.java:2394)
> ...
> at org.apache.hadoop.fs.s3a.S3AFileSystem.rename(S3AFileSystem.java:2392)
> at org.apache.spark.rdd.ReliableCheckpointRDD$.writePartitionToCheckpointFile(ReliableCheckpointRDD.scala:229)
> at org.apache.spark.rdd.ReliableCheckpointRDD$.$anonfun$writeRDDToCheckpointDirectory$1(ReliableCheckpointRDD.scala:168)
> at org.apache.spark.scheduler.Task.run(Task.scala:147)
> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:650)
> {noformat}
> followed by:
> {noformat}
> ERROR TaskSetManager: Task 379 in stage 105.0 failed 4 times; aborting job
> {noformat}
> Speculative kills were visible throughout the run; task 379.0 was still running when attempt 379.4 hit the already-committed part file.
> h2. Root cause
> {{writePartitionToCheckpointFile}} (master):
> {code:scala}
> if (!fs.rename(tempOutputPath, finalOutputPath)) {
> if (!fs.exists(finalOutputPath)) {
> logInfo(...)
> fs.delete(tempOutputPath, false)
> throw SparkCoreErrors.checkpointFailedToSaveError(ctx.attemptNumber(), finalOutputPath)
> } else {
> // Some other copy of this task must've finished before us and renamed it
> ...
> }
> }
> {code}
> The dest-exists race is only handled when {{rename()}} reports it by returning {{{}false{}}}. The Hadoop FileSystem specification does not guarantee that; HADOOP-16721 explicitly changed S3A "away from consistency with HDFS to one where applications find out what is wrong", raising {{FileAlreadyExistsException}} for a file destination. Structured Streaming's {{CheckpointFileManager}} was hardened for divergent rename semantics (e.g. SPARK-40826); the RDD checkpoint writer was not.
> h2. Reproduction
> # Configure an S3A (or ABFS) checkpoint dir, Hadoop >= 3.3.1, {{spark.speculation=true}} (aggressive settings help:
> {{{}spark.speculation.multiplier=1.0{}}}, {{{}spark.speculation.quantile=0.1{}}})
> # {{sc.setCheckpointDir("s3a://bucket/checkpoints")}}
> # Checkpoint an RDD with skewed/slow partitions so speculative attempts launch, e.g. a mapPartitions that sleeps on a subset of partitions; call {{rdd.checkpoint()}} then an action
> # A speculative attempt losing the rename race fails the task; all retries of that task then fail deterministically and the job aborts
> Also reproducible without speculation when a task fails *after* its rename commits and the task is retried (e.g. executor lost during post-write bookkeeping): the retry finds the destination and dies the same way.
> h2. Proposed fix
> Fold the exception into the existing false-return path, which already re-checks the destination defensively:
> {code:scala}
> val renamed = try {
> fs.rename(tempOutputPath, finalOutputPath)
> } catch {
> case _: FileAlreadyExistsException => false
> }
> if (!renamed) {
> // existing logic: if finalOutputPath exists, another attempt committed — clean up temp and
> // succeed; otherwise delete temp and throw checkpointFailedToSaveError
> }
> {code}
> Semantics are unchanged relative to HDFS: whichever attempt commits the part file wins, exactly as the winner-takes-all rename behaves today on filesystems that return {{{}false{}}}. I can submit a PR with this change and a test.
> h2. Workaround
> None good: speculation cannot be disabled per-job, so users must disable it application-wide, checkpoint to HDFS-semantics storage, or replace {{checkpoint()}} with a manual write/read-back through a file-source write (which goes through the output commit coordinator and is speculation-safe).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org