cloud-fan commented on code in PR #58167:
URL: https://github.com/apache/spark/pull/58167#discussion_r4139411707
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -83,6 +87,20 @@ private[spark] class ReliableCheckpointRDD[T: ClassTag](
throw SparkCoreErrors.invalidCheckpointDirectoryError(path,
expectedFileName)
}
}
+ // If a partition-count metadata file is present and the integrity check
is enabled,
+ // verify no trailing files are missing. Directories written by earlier
Spark versions
+ // have no such file; a missing file is silently tolerated for backward
compatibility.
+ // Set spark.checkpoint.verifyPartitionCount.enabled=false to suppress
this check when
+ // recovering a legitimately truncated directory. See SPARK-58883.
+ if (context.conf.get(CHECKPOINT_VERIFY_PARTITION_COUNT_ENABLED)) {
+ ReliableCheckpointRDD.readPartitionCountFromCheckpointDir(context,
checkpointPath)
+ .foreach { expected =>
+ if (inputFiles.length != expected) {
Review Comment:
**Non-blocking (P2):** The tests only cover this condition with fewer files
than recorded. A regression from `!=` to `<` would therefore leave all of them
green. Please add a case with a recorded count of 4 and contiguous parts
`part-00000` through `part-00004`, and assert the greater-than mismatch details.
**Verification:**
- **Regression:** A checkpoint with recorded count four and contiguous parts
zero through four is rejected with the expected greater-than mismatch details.
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -3167,6 +3167,20 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val CHECKPOINT_VERIFY_PARTITION_COUNT_ENABLED =
+ ConfigBuilder("spark.checkpoint.verifyPartitionCount.enabled")
+ .internal()
+ .doc("When true, a checkpoint directory written by a Spark version that
supports " +
Review Comment:
**Non-blocking (P2):** The writer publishes `_num_partitions` regardless of
this setting; only `getPartitions` checks the flag. An application can
therefore write with the flag disabled and later have the same metadata
enforced by a default-configured reader. Could this text state that publication
is always attempted best-effort and that the flag only bypasses the read-time
comparison?
**Verification:**
- **Inspection:** The configuration text agrees with the unconditional
writer call and the reader-only configuration branch for both enabled and
disabled values.
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -280,6 +327,122 @@ private[spark] object ReliableCheckpointRDD extends
Logging {
}
}
+ /**
+ * Write the partition count of the checkpointed RDD to the checkpoint
directory so that
+ * a later read via [[SparkContext.checkpointFile]] can detect a truncated
directory.
+ * The file is written atomically (temp path then rename) so a torn write
leaves no partial
+ * file, and the temp file is removed if publication fails at any point
after it is created;
+ * the payload layout is documented at [[PARTITION_COUNT_FORMAT_VERSION]].
+ * This is done on a best-effort basis; any exception is caught, logged and
ignored so that
+ * an inability to write the file does not prevent checkpointing. See
SPARK-58883.
+ */
+ private def writePartitionCountToCheckpointDir(
+ sc: SparkContext, partitionCount: Int, checkpointDirPath: Path): Unit = {
+ try {
+ val countFilePath = new Path(checkpointDirPath,
checkpointPartitionCountFileName())
+ val tmpFilePath = new Path(
+ checkpointDirPath, s".${checkpointPartitionCountFileName()}-tmp")
+ val bufferSize = sc.conf.get(BUFFER_SIZE)
+ val fs = countFilePath.getFileSystem(sc.hadoopConfiguration)
+ // A cleanup failure must not mask the publication failure being
handled, so it is only
+ // logged.
+ def deleteTempFile(): Unit = {
+ try {
+ fs.delete(tmpFilePath, false)
Review Comment:
**Non-blocking (P2):** `FileSystem.delete` can return `false` without
throwing when the temp file was not removed. Ignoring that result means a
post-create publication failure can leave `._num_partitions-tmp` behind without
the promised cleanup warning. Please check the Boolean result, warn when it is
false, and cover that failure signal in the injected-filesystem tests.
**Verification:**
- **Regression:** After a post-create publication failure, an injected false
delete result leaves checkpointing recoverable and emits both
publication-failure and cleanup-failure evidence.
- **Behavior:** A successful cleanup removes the temporary metadata path
without adding a cleanup warning.
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -69,6 +71,8 @@ private[spark] class ReliableCheckpointRDD[T: ClassTag](
* Since the original RDD may belong to a prior application, there is no way
to know a
* priori the number of partitions to expect. This method assumes that the
original set of
* checkpoint files are fully preserved in a reliable storage across
application lifespans.
+ * When the directory was written by a Spark version that supports
SPARK-58883, a
Review Comment:
**Non-blocking (P2):** A current writer can also omit this file because
publication is best-effort and any publication failure is tolerated. The reader
then reaches the same missing-file path, so describing absence as proof of an
older checkpoint is inaccurate. Please cover both legacy checkpoints and failed
best-effort publication in the documentation and diagnostic.
**Verification:**
- **Inspection:** Every changed statement about missing metadata is
consistent with the writer catch path that completes checkpointing without the
final metadata file.
- **Compatibility:** A missing metadata file remains non-authoritative and
recovery continues without partition-count enforcement.
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -1076,6 +1076,13 @@
],
"sqlState" : "58030"
},
+ "CHECKPOINT_TRUNCATED_DIRECTORY" : {
+ "message" : [
+ "The checkpoint directory <path> was written with <expected>
partition(s) but only <found> partition file(s) were found. One or more
trailing part files may have been lost.",
Review Comment:
**Non-blocking (P2):** This message is also used when `found > expected`.
For example, a valid count of 4 plus contiguous `part-00004` produces
`found=5`, yet the error says only five files were found and that files may
have been lost. Please use direction-neutral mismatch wording or distinguish
missing files from unexpected extra files.
**Verification:**
- **Behavior:** Valid metadata with fewer contiguous part files and valid
metadata with extra contiguous part files produce direction-appropriate
structured failures with the correct path and counts.
- **Compatibility:** Disabled verification and absent or invalid metadata
continue to bypass the new comparison exactly as before.
--
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]
