(spark) branch branch-4.3 updated: [SPARK-60264][SS] Propagate close() failures in AbortableStreamBasedCheckpointFileManager

[pt...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

peter-toth pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.3 by this push:
     new 21c9e37ec6b0 [SPARK-60264][SS] Propagate close() failures in AbortableStreamBasedCheckpointFileManager
21c9e37ec6b0 is described below

commit 21c9e37ec6b088e79b0963d91ad76932335cf011
Author: Steve Vaughan <em...@stevevaughan.me>
AuthorDate: Fri Oct 16 17:24:59 2026 +0200

    [SPARK-60264][SS] Propagate close() failures in AbortableStreamBasedCheckpointFileManager
    
    ### What changes were proposed in this pull request?
    
    `AbortableStreamBasedFSDataOutputStream.close()` now rethrows the exception after logging it, instead of swallowing it. When the destination already exists it now throws Hadoop's `FileAlreadyExistsException` (`org.apache.hadoop.fs`) instead of `java.nio.file`'s. `cancel()` is unchanged and stays best-effort, like `RenameBasedFSDataOutputStream.cancel()`.
    
    Tests:
    - A new abstract `AbortableStreamBasedCheckpointFileManagerTests` holds the tests that apply to any file system with abortable streams. Both `AbortableStreamBasedCheckpointFileManagerSuite` and `AwsS3AbortableStreamBasedCheckpointFileManagerSuite` extend it. The S3A suite used to extend the local suite; it now extends the base so it doesn't inherit local-only tests.
    - `SPARK-60264: close() should fail if the destination was created after createAtomic`: in the shared base, so it also runs against S3A.
    - `SPARK-60264: close() should propagate failures of the underlying stream`: local only. It removes the parent directory before `close()`. Like S3A, the test file system writes the file only when the stream is closed, so that write fails. It is local only because on S3A deleting the parent prefix does not make the upload fail.
    
    ### Why are the changes needed?
    
    With S3A, `close()` is where the object is uploaded (a single PUT, or completing the multipart upload). Because failures were swallowed, a failed upload of a checkpoint file looked like a successful write. For example, `HDFSMetadataLog.add()` returned `true` and cached the entry, and the query carried on without the file.
    
    If the lost file is `offsets/N` and batch N then completes, the checkpoint has `commits/N` but no `offsets/N`. `MicroBatchExecution.populateStartOffsets` has no branch for a commit log that is ahead of the offset log. So if the query stops before `offsets/N+1` is written, every restart resumes at batch N-1, re-runs that already-committed batch, and fails `commitLog.add(N-1)` with `[CONCURRENT_STREAM_LOG_UPDATE] Concurrent update to the log. Multiple streaming jobs detected for N-1.` S [...]
    
    The same `catch` also swallowed the `FileAlreadyExistsException` that `close()` throws when the destination appears between `createAtomic()` and `close()`. The write was then reported as successful even though the other writer's file was kept. That exception was also `java.nio.file`'s class, which `HDFSMetadataLog.write` does not map to its concurrent-query error.
    
    Swallowing contradicts the `CheckpointFileManager.createAtomic` contract and the behavior of the rename-based managers, whose `close()` propagates both kinds of failure. This is a follow-up to SPARK-40039 (apache/spark#37474, re-landed as apache/spark#37687), which introduced this class; the handling has not changed since.
    
    ### Does this PR introduce _any_ user-facing change?
    
    Yes, for queries that set `spark.sql.streaming.checkpointFileManagerClass` to `org.apache.spark.internal.io.cloud.AbortableStreamBasedCheckpointFileManager`. A failed write through this manager now throws instead of being logged and ignored:
    - Offset and commit logs, and other metadata logs: in micro-batch queries the batch fails and the query stops, and a restart resumes from the latest offsets file.
      - If the failed write was `offsets/N`, batch N had not run yet. With async progress tracking or the real-time trigger, offsets are written asynchronously or after the batch, so batch N may already have run.
      - If it was `commits/N`, batch N had already reached the sink, so it runs again on restart. Sinks that don't skip already-committed batch IDs write it twice. That is the same as with the default checkpoint file manager.
    - State store files written during a batch: the task fails and Spark retries it, instead of the file silently going missing.
    - A destination created by another writer between `createAtomic()` and `close()`: metadata logs now fail with their concurrent-query error instead of reporting a successful write. On S3A this check remains best-effort, because it checks for the object and then uploads.
    
    The success path is unchanged, so SPARK-40039's rename-free performance is unaffected.
    
    ### How was this patch tested?
    
    Added the two tests described above. In `AbortableStreamBasedCheckpointFileManagerSuite` both fail without this change ("no exception was thrown") and pass with it, and the existing test still passes. `AwsS3AbortableStreamBasedCheckpointFileManagerSuite` (an `IntegrationTestSuite` that needs `S3A_PATH` and is excluded by default) was not run.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Claude Code (Claude Opus 5.5)
    
    Closes #59518 from snmvaughan/feature/abortable-checkpoint-close-master.
    
    Authored-by: Steve Vaughan <em...@stevevaughan.me>
    Signed-off-by: Peter Toth <pe...@gmail.com>
    (cherry picked from commit f6eaea6cc0ef2b529429f8a16da6c632642c20a2)
    Signed-off-by: Peter Toth <pe...@gmail.com>
---
 ...AbortableStreamBasedCheckpointFileManager.scala |  2 +-
 ...ableStreamBasedCheckpointFileManagerSuite.scala | 57 ++++++++++++++++++++--
 2 files changed, 53 insertions(+), 6 deletions(-)

diff --git a/hadoop-cloud/src/main/scala/org/apache/spark/internal/io/cloud/AbortableStreamBasedCheckpointFileManager.scala b/hadoop-cloud/src/main/scala/org/apache/spark/internal/io/cloud/AbortableStreamBasedCheckpointFileManager.scala
index e64ba7882b91..3f9bd22bc3ad 100644
--- a/hadoop-cloud/src/main/scala/org/apache/spark/internal/io/cloud/AbortableStreamBasedCheckpointFileManager.scala
+++ b/hadoop-cloud/src/main/scala/org/apache/spark/internal/io/cloud/AbortableStreamBasedCheckpointFileManager.scala
@@ -16,7 +16,6 @@
  */
 package org.apache.spark.internal.io.cloud
 
-import java.nio.file.FileAlreadyExistsException
 import java.util.EnumSet
 
 import scala.util.control.NonFatal
@@ -74,6 +73,7 @@ class AbortableStreamBasedCheckpointFileManager(path: Path, hadoopConf: Configur
           case NonFatal(e) =>
             logWarning(log"Error closing ${MDC(LogKeys.PATH, path)} " +
               log"(stream: ${MDC(LogKeys.FS_DATA_OUTPUT_STREAM, fsDataOutputStream)})", e)
+            throw e
       } finally {
         terminated = true
       }
diff --git a/hadoop-cloud/src/test/scala/org/apache/spark/internal/io/cloud/AbortableStreamBasedCheckpointFileManagerSuite.scala b/hadoop-cloud/src/test/scala/org/apache/spark/internal/io/cloud/AbortableStreamBasedCheckpointFileManagerSuite.scala
index 0b7e65ff78d9..f04bfa2ccc3a 100644
--- a/hadoop-cloud/src/test/scala/org/apache/spark/internal/io/cloud/AbortableStreamBasedCheckpointFileManagerSuite.scala
+++ b/hadoop-cloud/src/test/scala/org/apache/spark/internal/io/cloud/AbortableStreamBasedCheckpointFileManagerSuite.scala
@@ -16,7 +16,7 @@
  */
 package org.apache.spark.internal.io.cloud
 
-import java.io.File
+import java.io.{File, IOException}
 
 import scala.util.Properties
 
@@ -30,9 +30,40 @@ import org.apache.spark.internal.io.cloud.abortable.AbortableFileSystem
 import org.apache.spark.sql.execution.streaming.CheckpointFileManagerTests
 import org.apache.spark.sql.execution.streaming.checkpointing.CheckpointFileManager
 
-class AbortableStreamBasedCheckpointFileManagerSuite
+/** Tests that apply to any file system with abortable stream support. */
+abstract class AbortableStreamBasedCheckpointFileManagerTests
   extends CheckpointFileManagerTests with Logging {
 
+  override def checkLeakingCrcFiles(path: Path): Unit = { }
+
+  test("SPARK-60264: close() should fail if the destination was created after createAtomic") {
+    withTempHadoopPath { basePath =>
+      val fm = createManager(basePath)
+      val dir = new Path(basePath, "concurrent-create")
+      fm.mkdirs(dir)
+      val path = new Path(dir, "file")
+      val out = fm.createAtomic(path, overwriteIfPossible = false)
+      out.writeInt(1)
+      // Another writer creates the destination between createAtomic() and close().
+      val concurrentOut = fm.createAtomic(path, overwriteIfPossible = true)
+      concurrentOut.writeInt(2)
+      concurrentOut.close()
+      intercept[FileAlreadyExistsException] {
+        out.close()
+      }
+      val in = fm.open(path)
+      try {
+        assert(in.readInt() === 2)
+      } finally {
+        in.close()
+      }
+    }
+  }
+}
+
+class AbortableStreamBasedCheckpointFileManagerSuite
+  extends AbortableStreamBasedCheckpointFileManagerTests {
+
   override def withTempHadoopPath(p: Path => Unit): Unit = {
     withTempDir { f: File =>
       val basePath = new Path(AbortableFileSystem.ABORTABLE_FS_SCHEME, null, f.getAbsolutePath)
@@ -40,19 +71,35 @@ class AbortableStreamBasedCheckpointFileManagerSuite
     }
   }
 
-  override def checkLeakingCrcFiles(path: Path): Unit = { }
-
   override def createManager(path: Path): CheckpointFileManager = {
     val conf = new Configuration()
     conf.set(s"fs.AbstractFileSystem.${AbortableFileSystem.ABORTABLE_FS_SCHEME}.impl",
       "org.apache.spark.internal.io.cloud.abortable.AbstractAbortableFileSystem")
     new AbortableStreamBasedCheckpointFileManager(path, conf)
   }
+
+  test("SPARK-60264: close() should propagate failures of the underlying stream") {
+    withTempHadoopPath { basePath =>
+      val fm = createManager(basePath)
+      val dir = new Path(basePath, "dir")
+      fm.mkdirs(dir)
+      val path = new Path(dir, "file")
+      val out = fm.createAtomic(path, overwriteIfPossible = false)
+      out.writeInt(1)
+      // Like S3A, the test file system only writes the file when the stream is closed, so
+      // removing the parent directory makes that write fail.
+      fm.delete(dir)
+      intercept[IOException] {
+        out.close()
+      }
+      assert(!fm.exists(path))
+    }
+  }
 }
 
 @IntegrationTestSuite
 class AwsS3AbortableStreamBasedCheckpointFileManagerSuite
-    extends AbortableStreamBasedCheckpointFileManagerSuite with BeforeAndAfter {
+    extends AbortableStreamBasedCheckpointFileManagerTests with BeforeAndAfter {
 
   val s3aPath = Properties.envOrNone("S3A_PATH")
 


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org