Steve Vaughan created SPARK-60264:
-------------------------------------
Summary: 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: 4.2.0, 4.0.0, 4.1.0, 3.5.0, 3.4.0, 4.3.0
Environment:
spark.sql.streaming.checkpointFileManagerClass=org.apache.spark.internal.io.cloud.AbortableStreamBasedCheckpointFileManager
(S3A)
Reporter: Steve Vaughan
{{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]
