dongjoon-hyun opened a new pull request, #59709: URL: https://github.com/apache/spark/pull/59709
### What changes were proposed in this pull request? This PR fixes two robustness issues in `FileSystemPersistenceEngine`, which backs the `FILESYSTEM` recovery mode of the standalone Master. 1. `serializeIntoFile` keeps the raw `FileOutputStream` as `rawOut` and closes the serialization stream, the codec stream and `rawOut`, in this order, each guaranteed by a nested `Utils.tryWithSafeFinally`. The close order from SPARK-46205, where the serializer and the codec flush before the file closes, is unchanged. The codec stream is now created inside the `tryWithSafeFinally` body (as SPARK-60419 does on the read side), so the raw stream is closed even when the codec stream constructor throws. 2. `serializeIntoFile` deletes the file and rethrows when writing fails after it created the file. It never deletes an existing file: `File already exists` and `Could not create file` are thrown before the file is created. 3. `read` skips a file that cannot be deserialized instead of failing as a whole. It logs a warning with the path and the exception, and renames the file to `corrupted_<name>`, which no longer matches the `app_`, `driver_` or `worker_` prefix and is kept for inspection. Items 2 and 3 together are the smallest design that unblocks recovery in all cases: - Deleting on write failures alone does not cover a crash or a power loss in the middle of a write, or files that released versions already left behind. - Skipping on read alone keeps a failed write's file under its real name until the next start, which then warns about it and renames it. - Writing to a temporary file, then `fsync` and `ATOMIC_MOVE`, would add an `fsync` to every app, driver and worker registration on the Master's RPC thread, which is slow on an NFS-mounted recovery directory. It also needs cleanup of leftover temporary files, `ATOMIC_MOVE` support from the file system, and a directory `fsync` that Java cannot do portably, and it still does not cover files that are already broken. Durability is out of the scope of this PR. Unlike `ZooKeeperPersistenceEngine`, which deletes unreadable entries, `read` renames them. This keeps the data when the cause is not a broken file, e.g. an incompatible class, a changed `spark.deploy.recoveryCompressionCodec`, or a transient I/O error. SPARK-60419 (https://github.com/apache/spark/pull/59670) fixed the double open in `deserializeFromFile`. This PR keeps that method as is, and fixes the write side and the per-file handling in `read`. ### Why are the changes needed? **Stream leaks on write failures.** When the last write fails during close, e.g. with `ENOSPC` or `EDQUOT`, `out.close()` throws. `ObjectOutputStream.close()` calls `flush()` first, which pushes the buffered data through the codec, so it never reaches the codec stream's `close()`. The `finally` block then skips `fileOut.close()`: - The raw `FileOutputStream` stays open until GC, with or without a codec. - With zstd, `ZstdOutputStreamNoFinalizer` is never closed, so its native `CStream` is never freed. The class has no finalizer as a fallback. Reaching `fileOut.close()` is not enough. When their final write fails, the `close()` methods of `LZ4BlockOutputStream`, `ZstdOutputStreamNoFinalizer` and the non-parallel `LZFOutputStream` throw before closing the stream they wrap, and nothing else referenced the raw `FileOutputStream`. The raw stream was also opened before the `try` block, so a codec stream constructor that throws leaked it. **Partially written files block recovery.** `serializeIntoFile` creates the file before writing it, never deletes it on failure, and does not `fsync`. `read` fails as a whole when any file cannot be deserialized. So an empty or truncated `app_`, `driver_` or `worker_` file, left by a failed write, a crash in the middle of a write, or a power loss, makes `readPersistedData` throw in the `ElectedLeader` handler on the next start. The RPC inbox only logs `Ignoring error`, the Master stays in `STANDBY`, and `MonarchyLeaderAgent` never elects it again. The Master is unusable until someone removes the file manually. `ZooKeeperPersistenceEngine` already logs, deletes and skips unreadable entries. ### Does this PR introduce _any_ user-facing change? Yes, in the `FILESYSTEM` recovery mode, compared with the released versions: - A failed persist no longer leaves an empty or truncated file in the recovery directory, and no longer leaks its file descriptor. - An unreadable file in the recovery directory no longer keeps the Master in `STANDBY`. The Master logs a warning, renames the file to `corrupted_<name>`, and recovers the other apps, drivers and workers. This also applies when the recovery directory was not cleared after changing `spark.deploy.recoveryMode` or `spark.deploy.recoveryCompressionCodec`, which the documentation asks users to do. ### How was this patch tested? Pass the CIs with the newly added test cases, which fail without the main code change: - `FileSystemPersistenceEngine closes and deletes the file when persist fails` passes a test-only `CompressionCodec` through the engine's `codec` constructor parameter. Its output stream fails on `flush()` and `close()` without closing the stream it wraps, as the lz4, zstd and lzf streams do on a full disk, or it fails to be created. The test checks that the codec stream's `close()` is called, that the raw `FileOutputStream` is closed (`getFD.valid()` is `false`), and that no file is left. Without the fix, the full-disk case fails with `false did not equal true` (the codec stream is never closed), and the constructor case fails with `codec.fileOut.getFD().valid() was true` (the raw stream leaks). - `FileSystemPersistenceEngine skips and renames unreadable files` writes an empty file and a truncated file next to a valid one. `read` returns only the valid object, the two files are renamed to `corrupted_<name>`, and their names can be persisted again. Without the fix, `read` throws `java.io.EOFException`. - The existing `SPARK-46191` test case also checks that a persist failing with `File already exists` keeps the existing file. ``` $ build/sbt 'core/testOnly *PersistenceEngineSuite *RecoverySuite' ... [info] Tests: succeeded 28, failed 0, canceled 0, ignored 0, pending 0 ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: 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]
