snmvaughan opened a new pull request, #59518:
URL: https://github.com/apache/spark/pull/59518
### 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.` Sinks that don't skip already-committed batch
IDs, for example a `foreachBatch` function that appends to a table, write that
batch again on each restart. If the query stops while `offsets/N+1` is the
latest offsets file, restarts fail with `batch N doesn't exist`.
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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
