[
https://issues.apache.org/jira/browse/SPARK-60264?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-60264:
-----------------------------------
Labels: pull-request-available (was: )
> AbortableStreamBasedCheckpointFileManager swallows close() failures, so
> failed checkpoint writes look successful
> ----------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-60264
> URL: https://issues.apache.org/jira/browse/SPARK-60264
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 3.4.0, 3.5.0, 4.1.0, 4.0.0, 4.2.0, 4.3.0
> Environment:
> spark.sql.streaming.checkpointFileManagerClass=org.apache.spark.internal.io.cloud.AbortableStreamBasedCheckpointFileManager
> (S3A)
> Reporter: Steve Vaughan
> Priority: Major
> Labels: pull-request-available
>
> {{AbortableStreamBasedCheckpointFileManager}} (hadoop-cloud, added in
> SPARK-40039 for S3A) writes each checkpoint file directly to its final path
> and relies on the output stream's {{close()}} to commit it. With S3A,
> {{close()}} is where the object is uploaded (a single PUT, or completion of a
> multipart upload).
> {{AbortableStreamBasedFSDataOutputStream.close()}} catches every {{NonFatal}}
> exception, logs {{Error closing <path>}} at WARN level and returns normally:
> # A failed upload is reported as a successful write.
> {{HDFSMetadataLog.add()}} returns {{true}} and caches the entry, so the query
> continues although, for example, {{offsets/N}} or {{commits/N}} was never
> written. State store files and other files written through {{createAtomic()}}
> are affected the same way. The missing file is noticed only later, for
> example on restart: if {{offsets/N}} is missing, {{commits/N}} exists and the
> query stops before {{offsets/N+1}} is written, every restart resumes at batch
> N-1, runs it again (sinks that don't skip already-committed batch IDs, such
> as {{foreachBatch}}, receive it again) and then fails with
> {{CONCURRENT_STREAM_LOG_UPDATE}}.
> # When {{overwriteIfPossible}} is false and another writer creates the
> destination after {{createAtomic()}}, {{close()}} aborts the upload and
> throws {{FileAlreadyExistsException}}, but the same catch swallows it, so the
> write is reported as successful while the other writer's file is kept. The
> exception is also {{java.nio.file.FileAlreadyExistsException}} instead of
> {{org.apache.hadoop.fs.FileAlreadyExistsException}}, the class that
> {{HDFSMetadataLog}} maps to its concurrent-query error.
> This contradicts the {{CheckpointFileManager.createAtomic()}} contract and
> differs from the rename-based managers, whose {{close()}} propagates both
> failures. {{close()}} should rethrow after logging and throw Hadoop's
> {{FileAlreadyExistsException}}; {{cancel()}} can stay best-effort.
> Reproduction: add this test to
> {{AbortableStreamBasedCheckpointFileManagerSuite}} (hadoop-cloud). Its test
> file system, like S3A, writes the file only in {{close()}}.
> {code:scala}
> test("close() failures are swallowed") {
> withTempHadoopPath { basePath =>
> val fm = createManager(basePath)
> val dir = new Path(basePath, "offsets")
> fm.mkdirs(dir)
> // 1. The write performed by close() fails.
> val file = new Path(dir, "0")
> val out = fm.createAtomic(file, overwriteIfPossible = false)
> out.writeInt(1)
> fm.delete(dir)
> out.close() // expected: IOException; actual: returns
> normally
> assert(!fm.exists(file)) // the file was never written
> // 2. Another writer creates the destination before close().
> fm.mkdirs(dir)
> val file2 = new Path(dir, "1")
> val first = fm.createAtomic(file2, overwriteIfPossible = false)
> first.writeInt(1)
> val second = fm.createAtomic(file2, overwriteIfPossible = true)
> second.writeInt(2)
> second.close()
> first.close() // expected: FileAlreadyExistsException;
> actual: returns normally
> }
> }
> {code}
> On master the test passes: both {{close()}} calls return normally, and the
> only trace is two {{WARN AbortableStreamBasedCheckpointFileManager: Error
> closing ...}} messages (with {{java.io.FileNotFoundException}} and
> {{java.nio.file.FileAlreadyExistsException}}).
> Related: SPARK-38033 (restart error when an offsets file is missing),
> SPARK-40826 (optional existence check after rename in the rename-based
> managers).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
