LuciferYang opened a new pull request, #58004: URL: https://github.com/apache/spark/pull/58004
### What changes were proposed in this pull request? This PR converts the five `_LEGACY_ERROR_TEMP_*` conditions in `SparkCoreErrors` that cover RDD checkpointing, continuing the cleanup under [SPARK-37935](https://issues.apache.org/jira/browse/SPARK-37935). Four get user-facing names; the fifth is an unreachable defensive branch and becomes an internal error. | Legacy | Builder | Now | SQLSTATE | |---|---|---|---| | `_LEGACY_ERROR_TEMP_3016` | `checkpointDirectoryHasNotBeenSetInSparkContextError` | `CHECKPOINT_DIRECTORY_NOT_SET` | 55019 | | `_3017` | `invalidCheckpointFileError` | `INVALID_CHECKPOINT_FILE` | 58030 | | `_3018` | `failToCreateCheckpointPathError` | `FAILED_CREATE_CHECKPOINT_DIRECTORY` | 58030 | | `_3019` | `checkpointRDDHasDifferentNumberOfPartitionsFromOriginalRDDError` | `CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH` | 58030 | | `_3020` | `mustSpecifyCheckpointDirError` | `INTERNAL_ERROR` (entry deleted) | XX000 | Four top-level names rather than one umbrella: `sqlState` lives on the umbrella, so grouping them would force a single SQLSTATE and lose the 55019/58030 split, and no umbrella sentence holds for all four. `_3016` fires before any job runs and is a user configuration mistake; the other three are storage failures during an action. `invalidCheckpointFileError` gains an `expectedFileName` parameter, taken from the `checkpointFileName(i)` the caller already computes, and reports the directory and file name separately. The old message named only the offending path, which is misleading here: the check walks the sorted `part-*` files and compares the *i*-th name against `part-%05d(i)`, so the path it reports is a perfectly valid file sitting where a missing one should be. With `part-00001` deleted from a 4-partition checkpoint the old message read `Invalid checkpoint file: .../part-00002` and sent the reader after the wrong file. `failToCreateCheckpointPathError`'s parameter is renamed `checkpointDirPath` to `path`, matching `INVALID_BUCKET_FILE` and the other path-carrying conditions. ### Reachability, per condition - **`_3016`, user configuration.** `RDD.checkpoint()` is public and `sc.checkpointDir` has exactly two writers: `SparkContext.setCheckpointDir` and, since 4.0.0, `spark.checkpoint.dir` (applied at `SparkContext.scala:614`). Calling `checkpoint()` with neither lands here. Reachable from `Dataset.checkpoint` too, which calls `internalRdd.checkpoint()`, so the message deliberately avoids naming RDDs. The old text mentioned only `setCheckpointDir`; the new one names the conf as well, which matters for a Connect client that cannot call the setter. - **`_3017`, incomplete checkpoint directory.** `getPartitions` requires the `part-*` files to be a contiguous `part-00000..part-000NN`. It fires on a partially written checkpoint, a manually pruned directory, or a directory handed to `SparkContext.checkpointFile` (which is how streaming recovery rebuilds `generatedRDDs`). Driver-side: `getPartitions` runs from `RDD.partitions`, and `partitions_` is `@transient`, so executors never compute it. - **`_3018`, storage refused the directory.** Only fires where a `FileSystem` reports failure by returning `false` from `mkdirs` rather than throwing, which is what HDFS and S3A can do; `LocalFileSystem` tends to throw instead. Driver-side, inside `RDD.doCheckpoint()` at the end of the first action. - **`_3019`, the write and the read-back disagree.** Not an engine invariant. The driver creates the directory, each executor writes its own `part-*` through its own `FileSystem`, and then the driver counts what its `FileSystem` lists, so the two sides of the comparison resolve in different JVMs. `setCheckpointDir` only *warns* when a cluster-mode application points at a local path, so a user who sets `/tmp/ckpt` on a cluster reaches this directly: the executors write to their own disks and the driver lists an empty directory. `getPartitions`' own scaladoc states the assumption being verified ("assumes that the original set of checkpoint files are fully preserved in a reliable storage"). Converting this one to `INTERNAL_ERROR` would report a storage misconfiguration as a Spark bug. - **`_3020`, unreachable.** `ReliableRDDCheckpointData`'s `cpDir` field throws when `sc.checkpointDir` is `None`, but its only construction site is `RDD.scala:1743`, two lines below the `context.checkpointDir.isEmpty` guard that raises `_3016`, inside the same `RDDCheckpointData.synchronized` block; `cpDir` is a plain `val`, evaluated there. The only writer that can store `None` is `setCheckpointDir(null)`, which no production code calls, and the field is `private[spark]`. It is reachable only by a cross-thread race that also makes the condition a duplicate of `_3016`, so it gets `internalError` rather than a second user-facing name for the same situation. ### Why are the changes needed? The error-conditions [README](https://github.com/apache/spark/blob/master/common/utils/src/main/resources/error/README.md) disallows new `_LEGACY_ERROR_TEMP_*` entries and asks existing ones to be resolved. This clears five of them. Three of the five were also weak on their own terms. `_3016` predates `spark.checkpoint.dir` and told the user about only one of the two ways to configure a directory. `_3017` pointed at the wrong file, as described above. `_3018` said only that creating the path failed, without saying that the filesystem reported it through a return value, which is the detail that tells an operator to look at permissions rather than at Spark. ### Does this PR introduce _any_ user-facing change? Yes, to error messages, with no API change. Converting any legacy condition changes the rendered string in two mechanical ways: `SparkThrowableHelper.formatErrorMessage` suppresses the `[CONDITION] ` prefix only for `_LEGACY_ERROR_`-prefixed names, and appends ` SQLSTATE: xxxxx` when a sqlState exists (legacy entries have none, so all four gain both). Beyond that: - `_3016`: `Checkpoint directory has not been set in the SparkContext` becomes `Cannot checkpoint because no checkpoint directory is configured. Set one with SparkContext.setCheckpointDir or the "spark.checkpoint.dir" configuration.` - `_3017`: `Invalid checkpoint file: <path>` becomes `Cannot read the checkpoint directory <path>: expected the partition file <expectedFileName> but found <fileName>. The partition files must be numbered contiguously from part-00000, one per partition.` The parameter set changes from one key to three, so `getMessageParameters()` gains `expectedFileName` and `fileName` while `path` narrows from the file to its directory. - `_3018`: `Failed to create checkpoint path <checkpointDirPath>` becomes `Failed to create the checkpoint directory <path> as FileSystem.mkdirs returned false.` The parameter is renamed, so `getMessageParameters()` has `path` where it had `checkpointDirPath`. - `_3019`: the three-line `Checkpoint RDD has a different number of partitions from original RDD. Original RDD [ID: ..., num of partitions: ...]; Checkpoint RDD [ID: ..., num of partitions: ...].` becomes `The checkpoint of RDD <originalRDDId> has <newRDDLength> partition(s), but the RDD itself has <originalRDDLength>. The checkpoint RDD is <newRDDId>.` plus a second line naming the two usual causes. Same four parameters. - `_3020` renders as an internal error. This does not change any job-failure message shape: the throw happens on the driver inside `RDD.checkpoint()`, before any task runs, so `DAGScheduler.abortStage`'s `isInternalError` filter is not involved. ### How was this patch tested? None of the five had any test coverage, and `CheckpointSuite` contained no `intercept` at all. Three tests are added to `CheckpointStorageSuite`, each asserting the condition and the SQLSTATE, and each failing against the pre-change code on both fields (the legacy entries carry no sqlState): - `"checkpoint() without a checkpoint directory"` builds a context with no checkpoint directory and asserts `CHECKPOINT_DIRECTORY_NOT_SET`. - `"reading a checkpoint directory with a missing partition file"` checkpoints a 4-partition RDD, deletes `part-00001`, then reads the directory back through `SparkContext.checkpointFile` and asserts `INVALID_CHECKPOINT_FILE` with all three message parameters, pinning that the reported expectation is `part-00001` and the file found is `part-00002`. - `"checkpoint path that cannot be created"` registers a `LocalFileSystem` subclass whose `mkdirs` returns `false` for the per-RDD directory and asserts `FAILED_CREATE_CHECKPOINT_DIRECTORY`. Two details are load-bearing: `LocalFileSystem` throws `FileAlreadyExistsException` when the path is occupied instead of returning `false`, so the failure has to be injected rather than staged on disk; and Hadoop caches `FileSystem` instances per scheme, so the test also sets `fs.file.impl.disable.cache=true` or an earlier test's real `LocalFileSystem` is used instead. `CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH` has no triggering test. Reproducing it means making the driver and the executors see different contents for the same directory, which `local` mode cannot do since both sides share one filesystem. A fake `FileSystem` that under-reports `listStatus` would exercise the assertion but would be testing the fake rather than the failure, so this condition is covered by inspection only. The SQLSTATE assertions were confirmed to be live by temporarily setting `CHECKPOINT_DIRECTORY_NOT_SET`'s value to `42000` and watching the test fail with `sqlState: expected '55019' but got '42000'` before restoring it. `checkError` skips the comparison when `sqlState` is `None` and `SparkThrowableSuite` only checks that a state is registered, so a wrong SQLSTATE would otherwise ship green. Ran `core/testOnly org.apache.spark.SparkThrowableSuite org.apache.spark.CheckpointSuite org.apache.spark.CheckpointStorageSuite` (68 tests, all passing). ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8) -- 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]
