cloud-fan commented on code in PR #58167:
URL: https://github.com/apache/spark/pull/58167#discussion_r4131129874
##########
core/src/test/scala/org/apache/spark/CheckpointSuite.scala:
##########
@@ -771,6 +774,207 @@ class CheckpointStorageSuite extends SparkFunSuite with
LocalSparkContext {
parameters = Map("path" -> rddPath.toString))
}
}
+
+ // SPARK-58883: truncated checkpoint directory (trailing part-* file
deleted) should be
+ // detected when reading back via SparkContext.checkpointFile.
+ test("SPARK-58883: reading a truncated checkpoint directory throws an
error") {
+ withTempDir { checkpointDir =>
+ val conf = new SparkConf().set(UI_ENABLED.key, "false")
+ sc = new SparkContext("local", "test", conf)
+ sc.setCheckpointDir(checkpointDir.toString)
+ val rdd = sc.makeRDD(1 to 20, numSlices = 4)
+ rdd.checkpoint()
+ rdd.collect()
+
+ val checkpointPath = new Path(rdd.getCheckpointFile.get)
+ val fs = checkpointPath.getFileSystem(sc.hadoopConfiguration)
+
+ // Delete the last partition file; the remaining files are still
contiguous so the
+ // old contiguity check would pass silently and return a 3-partition RDD.
+ val lastPartFile = new Path(checkpointPath, "part-00003")
+ assert(fs.exists(lastPartFile), "expected part-00003 to exist before
deletion")
+ fs.delete(lastPartFile, false)
+
+ // Reading back should now throw because _num_partitions records the
original count.
+ val recoveredRDD = sc.checkpointFile[Int](rdd.getCheckpointFile.get)
+ checkError(
+ exception = intercept[SparkException](recoveredRDD.partitions),
+ condition = "CHECKPOINT_TRUNCATED_DIRECTORY",
+ sqlState = Some("58030"),
+ parameters = Map(
+ "path" -> rdd.getCheckpointFile.get,
+ "expected" -> "4",
+ "found" -> "3"))
+
+ // With the config disabled, the same truncated directory should load
silently.
+ sc.conf.set(CHECKPOINT_VERIFY_PARTITION_COUNT_ENABLED, false)
+ try {
+ val suppressedRDD = sc.checkpointFile[Int](rdd.getCheckpointFile.get)
+ assert(suppressedRDD.partitions.length === 3)
+ } finally {
+ sc.conf.set(CHECKPOINT_VERIFY_PARTITION_COUNT_ENABLED, true)
+ }
+ }
+ }
+
+ test("SPARK-58883: checkpoint directory without _num_partitions is read
without error") {
+ // Backward compatibility: a checkpoint written before SPARK-58883 has no
_num_partitions
+ // file. Removing it must not prevent the RDD from being read.
+ withTempDir { checkpointDir =>
+ val conf = new SparkConf().set(UI_ENABLED.key, "false")
+ sc = new SparkContext("local", "test", conf)
+ sc.setCheckpointDir(checkpointDir.toString)
+ val rdd = sc.makeRDD(1 to 20, numSlices = 4)
+ rdd.checkpoint()
+ rdd.collect()
+
+ val checkpointPath = new Path(rdd.getCheckpointFile.get)
+ val fs = checkpointPath.getFileSystem(sc.hadoopConfiguration)
+
+ // Remove the metadata file to simulate a pre-SPARK-58883 checkpoint.
+ val countFile = new Path(checkpointPath, "_num_partitions")
+ assert(fs.exists(countFile), "expected _num_partitions to exist before
deletion")
+ fs.delete(countFile, false)
+
+ // Must recover normally with all 4 partitions and correct data.
+ val recovered = sc.checkpointFile[Int](rdd.getCheckpointFile.get)
+ assert(recovered.partitions.length === 4)
+ assert(recovered.collect().sorted === (1 to 20).toArray)
+ }
+ }
+
+ test("SPARK-58883: corrupted _num_partitions file is tolerated") {
+ withTempDir { checkpointDir =>
+ val conf = new SparkConf().set(UI_ENABLED.key, "false")
+ sc = new SparkContext("local", "test", conf)
+ sc.setCheckpointDir(checkpointDir.toString)
+ val rdd = sc.makeRDD(1 to 20, numSlices = 4)
+ rdd.checkpoint()
+ rdd.collect()
+
+ val checkpointPath = new Path(rdd.getCheckpointFile.get)
+ val fs = checkpointPath.getFileSystem(sc.hadoopConfiguration)
+
+ val countFile = new Path(checkpointPath, "_num_partitions")
+ assert(fs.exists(countFile), "expected _num_partitions to exist before
corruption")
+ val originalBytes = readAllBytes(fs, countFile)
+ // Format version 1: 1-byte version, 4-byte count, 8-byte CRC32 of the
first 5 bytes.
+ assert(originalBytes.length === 13)
Review Comment:
**Non-blocking (P2):** This assertion pins only the total length; every
subsequent case starts from bytes emitted by the same writer and is interpreted
by the paired reader. A coordinated change to the version byte, count
endianness, or stored CRC can remain green without a version bump, and
checksum-valid unsupported-version or negative-count payloads do not exercise
their rejection branches.
**Verification:**
- **Compatibility:** Assert the exact semantic fields of a version-1
partition-count payload using an independently computed expected encoding.
- **Regression:** Construct checksum-valid unsupported-version and
negative-count payloads and verify the inactive-check warning and successful
recovery with the actual checkpoint partitions.
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -280,6 +327,103 @@ 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; 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)
+ // Write to a temp path first so readers never see a partial file.
+ val fileOutputStream = fs.create(tmpFilePath, true, bufferSize)
+ val dos = new DataOutputStream(fileOutputStream)
+ Utils.tryWithSafeFinally {
+ val header = partitionCountHeader(partitionCount)
+ dos.write(header)
+ dos.writeLong(partitionCountChecksum(header))
+ } {
+ dos.close()
+ }
+ if (!fs.rename(tmpFilePath, countFilePath)) {
+ fs.delete(tmpFilePath, false)
+ logWarning(
+ log"Failed to rename ${MDC(TEMP_OUTPUT_PATH, tmpFilePath)} to " +
+ log"${MDC(PATH, countFilePath)}, " +
+ log"truncation detection will be inactive for this directory")
+ } else {
+ logDebug(s"Written partition count $partitionCount to $countFilePath")
+ }
+ } catch {
Review Comment:
**Non-blocking (P2):** After `fs.create` succeeds, `dos.write`, `dos.close`,
or `fs.rename` can throw into this catch. It logs and returns without deleting
`._num_partitions-tmp`, leaving an orphan artifact on a deliberately tolerated
failure path and violating the stated no-partial-file contract; the current
create-throw and rename-false cases do not exercise this path.
**Verification:**
- **Regression:** Inject at least one metadata publication exception after
the temp path exists and verify successful checkpoint completion, the
inactive-check warning, absence of both metadata paths, and successful recovery.
- **Behavior:** Verify existing successful publication and false-return
rename behavior remain unchanged while cleanup failures do not mask the
original publication failure.
--
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]
