This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 5ef3a9492f9e [SPARK-58770][CORE] Assign a name to the error condition
`_LEGACY_ERROR_TEMP_3016-3020`
5ef3a9492f9e is described below
commit 5ef3a9492f9ef4a412ab21aa5e4ba4eb43bc0b99
Author: YangJie <[email protected]>
AuthorDate: Fri Aug 21 10:51:48 2026 +0800
[SPARK-58770][CORE] Assign a name to the error condition
`_LEGACY_ERROR_TEMP_3016-3020`
### 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` | `invalidCheckpointDirectoryError` |
`INVALID_CHECKPOINT_DIRECTORY` | 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.
`invalidCheckpointDirectoryError` 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-parti [...]
`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` is set through `SparkContext.setCheckpointDir`, reachable
either by calling it directly or, since 4.0.0, through `spark.checkpoint.dir`,
which `SparkContext.scala:623` forwards to the same setter. 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 [...]
- **`_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 dis [...]
- **`_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. `mustSpecifyCheckpointDirError` has
a single call site, the `getOrElse` on that field. It is reachable only by a
cross-thread race that also make [...]
### 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.` 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`: `Checkpoint dir must be specified.` becomes `[INTERNAL_ERROR]
SparkContext.checkpointDir is unset when creating ReliableRDDCheckpointData.
SQLSTATE: XX000`, and `getMessageParameters()` goes from empty to `{"message":
...}` since `INTERNAL_ERROR`'s template is `<message>`. 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_DIRECTORY` 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 [...]
`CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH` has no triggering test. The
comparison it guards needs an original RDD to count against, and on the
read-back path there is none: `getPartitions`' own scaladoc notes there is "no
way to know a priori the number of partitions to expect". Reproducing the
write-path case 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. Two related gaps in t [...]
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)
Closes #58004 from LuciferYang/assign-name-legacy-3016-3020.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 53 ++++++++---------
.../org/apache/spark/errors/SparkCoreErrors.scala | 27 +++++----
.../apache/spark/rdd/ReliableCheckpointRDD.scala | 5 +-
.../scala/org/apache/spark/CheckpointSuite.scala | 69 +++++++++++++++++++++-
4 files changed, 114 insertions(+), 40 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index df1a1f3a5b8f..aa492d647f12 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -999,6 +999,13 @@
},
"sqlState" : "XX000"
},
+ "CHECKPOINT_DIRECTORY_NOT_SET" : {
+ "message" : [
+ "Cannot checkpoint because no checkpoint directory is configured.",
+ "Set one with `SparkContext.setCheckpointDir` or the
\"spark.checkpoint.dir\" configuration."
+ ],
+ "sqlState" : "55019"
+ },
"CHECKPOINT_FILE_CHECKSUM_VERIFICATION_FAILED" : {
"message" : [
"Checksum verification failed, the file may be corrupted. File:
<fileName>",
@@ -1014,6 +1021,13 @@
],
"sqlState" : "56000"
},
+ "CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH" : {
+ "message" : [
+ "The checkpoint of RDD <originalRDDId> has <newRDDLength> partition(s),
but the RDD itself has <originalRDDLength>. The checkpoint RDD is <newRDDId>.",
+ "This usually means the checkpoint directory is not on storage that both
the driver and the executors can read and write, or that some checkpoint files
did not survive until the checkpoint was read back."
+ ],
+ "sqlState" : "58030"
+ },
"CHECK_CONSTRAINT_VIOLATION" : {
"message" : [
"CHECK constraint <constraintName> <expression> violated by row with
values:",
@@ -2678,6 +2692,12 @@
],
"sqlState" : "42822"
},
+ "FAILED_CREATE_CHECKPOINT_DIRECTORY" : {
+ "message" : [
+ "Failed to create the checkpoint directory <path> as FileSystem.mkdirs
returned false."
+ ],
+ "sqlState" : "58030"
+ },
"FAILED_EXECUTE_UDF" : {
"message" : [
"User defined function (<functionName>: (<signature>) => <result>)
failed due to: <reason>."
@@ -3774,6 +3794,12 @@
},
"sqlState" : "42K03"
},
+ "INVALID_CHECKPOINT_DIRECTORY" : {
+ "message" : [
+ "Cannot read the checkpoint directory <path>: expected the partition
file <expectedFileName> but found <fileName>. The partition files must be
numbered contiguously from part-00000."
+ ],
+ "sqlState" : "58030"
+ },
"INVALID_CLONE_SESSION_REQUEST" : {
"message" : [
"Invalid session clone request."
@@ -11594,33 +11620,6 @@
"countByValueApprox() does not support arrays"
]
},
- "_LEGACY_ERROR_TEMP_3016" : {
- "message" : [
- "Checkpoint directory has not been set in the SparkContext"
- ]
- },
- "_LEGACY_ERROR_TEMP_3017" : {
- "message" : [
- "Invalid checkpoint file: <path>"
- ]
- },
- "_LEGACY_ERROR_TEMP_3018" : {
- "message" : [
- "Failed to create checkpoint path <checkpointDirPath>"
- ]
- },
- "_LEGACY_ERROR_TEMP_3019" : {
- "message" : [
- "Checkpoint RDD has a different number of partitions from original RDD.
Original",
- "RDD [ID: <originalRDDId>, num of partitions: <originalRDDLength>];",
- "Checkpoint RDD [ID: <newRDDId>, num of partitions: <newRDDLength>]."
- ]
- },
- "_LEGACY_ERROR_TEMP_3020" : {
- "message" : [
- "Checkpoint dir must be specified."
- ]
- },
"_LEGACY_ERROR_TEMP_3028" : {
"message" : [
"<errorMsg>"
diff --git a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
index cf931707c89e..442d4623c4a9 100644
--- a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
+++ b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
@@ -157,22 +157,30 @@ private[spark] object SparkCoreErrors {
def checkpointDirectoryHasNotBeenSetInSparkContextError(): Throwable = {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3016", messageParameters = Map.empty,
cause = null
+ errorClass = "CHECKPOINT_DIRECTORY_NOT_SET",
+ messageParameters = Map.empty,
+ cause = null
)
}
- def invalidCheckpointFileError(path: Path): Throwable = {
+ def invalidCheckpointDirectoryError(
+ partitionFilePath: Path,
+ expectedFileName: String): Throwable = {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3017",
- messageParameters = Map("path" -> s"$path"),
+ errorClass = "INVALID_CHECKPOINT_DIRECTORY",
+ messageParameters = Map(
+ "path" -> s"${partitionFilePath.getParent}",
+ "expectedFileName" -> expectedFileName,
+ "fileName" -> partitionFilePath.getName
+ ),
cause = null
)
}
def failToCreateCheckpointPathError(checkpointDirPath: Path): Throwable = {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3018",
- messageParameters = Map("checkpointDirPath" -> s"$checkpointDirPath"),
+ errorClass = "FAILED_CREATE_CHECKPOINT_DIRECTORY",
+ messageParameters = Map("path" -> s"$checkpointDirPath"),
cause = null
)
}
@@ -183,7 +191,7 @@ private[spark] object SparkCoreErrors {
newRDDId: Int,
newRDDLength: Int): Throwable = {
new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3019",
+ errorClass = "CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH",
messageParameters = Map(
"originalRDDId" -> s"$originalRDDId",
"originalRDDLength" -> s"$originalRDDLength",
@@ -200,9 +208,8 @@ private[spark] object SparkCoreErrors {
}
def mustSpecifyCheckpointDirError(): Throwable = {
- new SparkException(
- errorClass = "_LEGACY_ERROR_TEMP_3020", messageParameters = Map.empty,
cause = null
- )
+ SparkException.internalError(
+ "SparkContext.checkpointDir is unset when creating
ReliableRDDCheckpointData.")
}
def askStandaloneSchedulerToShutDownExecutorsError(e: Exception): Throwable
= {
diff --git
a/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala
b/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala
index fd42cea795d6..156a94abd628 100644
--- a/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala
@@ -78,8 +78,9 @@ private[spark] class ReliableCheckpointRDD[T: ClassTag](
.sortBy(_.getName.stripPrefix("part-").toInt)
// Fail fast if input files are invalid
inputFiles.zipWithIndex.foreach { case (path, i) =>
- if (path.getName != ReliableCheckpointRDD.checkpointFileName(i)) {
- throw SparkCoreErrors.invalidCheckpointFileError(path)
+ val expectedFileName = ReliableCheckpointRDD.checkpointFileName(i)
+ if (path.getName != expectedFileName) {
+ throw SparkCoreErrors.invalidCheckpointDirectoryError(path,
expectedFileName)
}
}
Array.tabulate(inputFiles.length)(i => new CheckpointRDDPartition(i))
diff --git a/core/src/test/scala/org/apache/spark/CheckpointSuite.scala
b/core/src/test/scala/org/apache/spark/CheckpointSuite.scala
index 58512a2282ac..976e9b94d23c 100644
--- a/core/src/test/scala/org/apache/spark/CheckpointSuite.scala
+++ b/core/src/test/scala/org/apache/spark/CheckpointSuite.scala
@@ -21,7 +21,7 @@ import java.io.File
import scala.reflect.ClassTag
-import org.apache.hadoop.fs.Path
+import org.apache.hadoop.fs.{LocalFileSystem, Path}
import
org.apache.spark.internal.config.CACHE_CHECKPOINT_PREFERRED_LOCS_EXPIRE_TIME
import org.apache.spark.internal.config.UI._
@@ -684,4 +684,71 @@ class CheckpointStorageSuite extends SparkFunSuite with
LocalSparkContext {
assert(flatMappedRDD.collect() === result)
}
}
+
+ test("checkpoint() without a checkpoint directory") {
+ sc = new SparkContext("local", "test", new SparkConf().set(UI_ENABLED.key,
"false"))
+ checkError(
+ exception = intercept[SparkException](sc.makeRDD(1 to 4).checkpoint()),
+ condition = "CHECKPOINT_DIRECTORY_NOT_SET",
+ sqlState = Some("55019"))
+ }
+
+ test("reading a checkpoint directory with a missing partition file") {
+ withTempDir { checkpointDir =>
+ sc = new SparkContext("local", "test", new
SparkConf().set(UI_ENABLED.key, "false"))
+ sc.setCheckpointDir(checkpointDir.toString)
+ val rdd = sc.makeRDD(1 to 20, numSlices = 4)
+ rdd.checkpoint()
+ assert(rdd.collect().toSeq === (1 to 20))
+
+ // Drop a middle partition file so the remaining names are no longer
part-00000..part-0000N.
+ val checkpointPath = new Path(rdd.getCheckpointFile.get)
+ val fs = checkpointPath.getFileSystem(sc.hadoopConfiguration)
+ assert(fs.delete(new Path(checkpointPath, "part-00001"), false))
+
+ val recovered = sc.checkpointFile[Int](checkpointPath.toString)
+ checkError(
+ exception = intercept[SparkException](recovered.partitions),
+ condition = "INVALID_CHECKPOINT_DIRECTORY",
+ sqlState = Some("58030"),
+ parameters = Map(
+ "path" -> checkpointPath.toString,
+ "expectedFileName" -> "part-00001",
+ "fileName" -> "part-00002"))
+ }
+ }
+
+ test("checkpoint path that cannot be created") {
+ withTempDir { checkpointDir =>
+ // MkdirsFailingFilesystem refuses to create the per-RDD directory and
reports it the way
+ // HDFS and S3A do, by returning false rather than throwing.
+ val conf = new SparkConf()
+ .set("spark.hadoop.fs.file.impl",
classOf[MkdirsFailingFilesystem].getName)
+ .set("spark.hadoop.fs.file.impl.disable.cache", "true")
+ .set(UI_ENABLED.key, "false")
+ sc = new SparkContext("local", "test", conf)
+ sc.setCheckpointDir(checkpointDir.toString)
+ val rdd = sc.makeRDD(1 to 4)
+ val rddPath = ReliableRDDCheckpointData.checkpointPath(sc, rdd.id).get
+
+ rdd.checkpoint()
+ checkError(
+ exception = intercept[SparkException](rdd.collect()),
+ condition = "FAILED_CREATE_CHECKPOINT_DIRECTORY",
+ sqlState = Some("58030"),
+ parameters = Map("path" -> rddPath.toString))
+ }
+ }
+}
+
+/**
+ * A filesystem that refuses to create a per-RDD checkpoint directory,
returning false from
+ * `mkdirs` instead of throwing. `LocalFileSystem` throws
`FileAlreadyExistsException` when the
+ * path is taken, while HDFS and S3A can report the failure through the return
value, which is
+ * what `ReliableCheckpointRDD` checks.
+ */
+class MkdirsFailingFilesystem extends LocalFileSystem {
+ override def mkdirs(f: Path): Boolean = {
+ if (f.getName.startsWith("rdd-")) false else super.mkdirs(f)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
