anshulbaliga7 commented on code in PR #58167:
URL: https://github.com/apache/spark/pull/58167#discussion_r3862080181
##########
core/src/test/scala/org/apache/spark/CheckpointSuite.scala:
##########
@@ -739,6 +739,92 @@ 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.
+ // The ReliableCheckpointRDD has no separate "original" RDD on the read
path, so
+ // its own id is used for both RDD id fields.
+ val recoveredRDD = sc.checkpointFile[Int](rdd.getCheckpointFile.get)
+ checkError(
+ exception = intercept[SparkException](recoveredRDD.partitions),
+ condition = "CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH",
+ sqlState = Some("58030"),
+ parameters = Map(
+ "originalRDDId" -> recoveredRDD.id.toString,
+ "originalRDDLength" -> "4",
+ "newRDDId" -> recoveredRDD.id.toString,
+ "newRDDLength" -> "3"))
+ }
+ }
+
+ 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")
+ fs.delete(countFile, false)
Review Comment:
Tests 2 and 3 didn't prove what their comments claimed so added
`assert(fs.exists(...))` preconditions to both, and test 3 now asserts the WARN
fires via withLogAppender, so it actually distinguishes "corruption tolerated"
from "the check never ran."
--
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]
