LuciferYang commented on code in PR #58167:
URL: https://github.com/apache/spark/pull/58167#discussion_r4237061116
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -170,6 +219,7 @@ private[spark] object ReliableCheckpointRDD extends Logging
{
if (originalRDD.partitioner.nonEmpty) {
writePartitionerToCheckpointDir(sc, originalRDD.partitioner.get,
checkpointDirPath)
}
+ writePartitionCountToCheckpointDir(sc, originalRDD.partitions.length,
checkpointDirPath)
Review Comment:
The count written at `:222` is `originalRDD.partitions.length`, and the
read-back at `:230` runs the new check in `getPartitions` first against that
same number. So on the write path the new `MISSING_FILES` pre-empts
`CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH` at `:231`, which is now reachable
only when `_num_partitions` was not written or not read back, or the check is
disabled. The usual way to hit this is a cluster-mode checkpoint dir on local
disk. That case used to get the SPARK-58770 hint that the directory is not on
storage both the driver and the executors can reach; now it says trailing files
may have been lost and suggests turning the flag off. Doing so only brings back
the old error at `:231`, and leaves detection off.
Moving `:222` after the check at `:230` fixes this: record N only once the
directory is known to hold N part files, so the read-back finds no
`_num_partitions` and skips the new check. A `LocalFileSystem` subclass whose
`listStatus` omits the last part file can test it by asserting the old error
condition.
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -83,6 +89,21 @@ private[spark] class ReliableCheckpointRDD[T: ClassTag](
throw SparkCoreErrors.invalidCheckpointDirectoryError(path,
expectedFileName)
}
}
+ // If a partition-count metadata file is present and the check is enabled,
verify the
+ // recorded count against the files found: fewer means trailing files were
lost, more means
+ // files were added that the checkpoint did not write. A missing file
(older Spark version
+ // or failed best-effort write) is tolerated. Set
+ // spark.checkpoint.verifyPartitionCount.enabled=false to skip only this
comparison when
+ // the directory must be loaded as-is. See SPARK-58883.
+ if (context.conf.get(CHECKPOINT_VERIFY_PARTITION_COUNT_ENABLED)) {
Review Comment:
The new flag does not take effect on DStream recovery, which is the only
production caller of `checkpointFile`. On `getOrCreate`, the SparkContext is
built from `Checkpoint.createSparkConf()`, which starts from the conf saved in
the checkpoint and re-reads only a fixed set of keys, `propertiesToReload`,
from the current environment. I checked this in spark-shell on 4.1.1 (that code
is unchanged since): restarting with `--conf
spark.checkpoint.verifyPartitionCount.enabled=false` leaves the key unset in
the recovered conf, while `spark.master`, which is on that list, picks up the
new value. So an operator who follows the error message and restarts with the
flag off gets the same error again.
Adding `CHECKPOINT_VERIFY_PARTITION_COUNT_ENABLED.key` to
`propertiesToReload` in `Checkpoint.scala` would fix it; `UI_FILTERS.key` in
that list already references a core config the same way. The streaming
`CheckpointSuite` test "recovery of conf through checkpoints" already drives
`createSparkConf()` through system properties, so one more case there would pin
it.
##########
core/src/test/scala/org/apache/spark/CheckpointSuite.scala:
##########
@@ -771,6 +775,313 @@ 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_DIRECTORY_PARTITION_COUNT_MISMATCH.MISSING_FILES",
+ 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: reading a checkpoint directory with extra partition files
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)
+
+ // Add a contiguous part-00004 that the checkpoint never wrote. The
contiguity check
+ // still passes, so only the recorded count can reject it, and the
mismatch must be
+ // reported as extra files rather than as truncation. A regression from
!= to < in the
+ // comparison would leave this directory accepted with 5 partitions.
+ val extraPartFile = new Path(checkpointPath, "part-00004")
+ assert(!fs.exists(extraPartFile), "expected part-00004 not to exist
before the copy")
+ writeAllBytes(fs, extraPartFile, readAllBytes(fs, new
Path(checkpointPath, "part-00003")))
+
+ val recoveredRDD = sc.checkpointFile[Int](rdd.getCheckpointFile.get)
+ checkError(
+ exception = intercept[SparkException](recoveredRDD.partitions),
+ condition =
"CHECKPOINT_DIRECTORY_PARTITION_COUNT_MISMATCH.UNEXPECTED_FILES",
+ sqlState = Some("58030"),
+ parameters = Map(
+ "path" -> rdd.getCheckpointFile.get,
+ "expected" -> "4",
+ "found" -> "5"))
+
+ // With the config disabled, all 5 part files load as-is.
+ sc.conf.set(CHECKPOINT_VERIFY_PARTITION_COUNT_ENABLED, false)
+ try {
+ val suppressedRDD = sc.checkpointFile[Int](rdd.getCheckpointFile.get)
+ assert(suppressedRDD.partitions.length === 5)
+ } 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)
Review Comment:
`:862` is the backward-compatibility test for directories without the file,
but it only checks the partition count and the data. If the
`FileNotFoundException` case at `ReliableCheckpointRDD.scala:445` is removed,
the exception falls into the `NonFatal` case at `:449`, the reader still
returns `None`, and the test stays green, while every read of a pre-upgrade
directory logs a "truncation detection will be inactive" WARN with a stack
trace.
Wrapping the read in `withLogAppender` and asserting that WARN is absent
would pin it.
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -280,6 +330,128 @@ 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, whether reported by a false return value or by an exception.
+ def deleteTempFile(): Unit = {
+ try {
+ if (!fs.delete(tmpFilePath, false)) {
+ logWarning(log"Failed to delete ${MDC(TEMP_OUTPUT_PATH,
tmpFilePath)}, " +
+ log"it may be left behind in the checkpoint directory")
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to delete ${MDC(TEMP_OUTPUT_PATH,
tmpFilePath)}, " +
+ log"it may be left behind in the checkpoint directory", e)
+ }
+ }
+ // Write to a temp path first so readers never see a partial file.
+ val fileOutputStream = fs.create(tmpFilePath, true, bufferSize)
+ // From here on the temp file exists and must not be left behind on any
failure.
+ val renamed = try {
+ val dos = new DataOutputStream(fileOutputStream)
+ Utils.tryWithSafeFinally {
+ val header = partitionCountHeader(partitionCount)
+ dos.write(header)
+ dos.writeLong(partitionCountChecksum(header))
+ } {
+ dos.close()
+ }
+ fs.rename(tmpFilePath, countFilePath)
Review Comment:
The reader accepts exactly 13 bytes, so any prefix left by a torn write
fails in `readFully` or `readLong` with `EOFException` and ends up where a
missing file does: detection is off for that directory. Temp-then-rename only
changes whether that read logs a WARN or a debug line. I asked for it in the
first round together with the fixed length, but the length alone already
exposes a torn write.
Since you asked whether to trim things back, this is the part I would trim:
write `_num_partitions` directly and keep the current WARN on failure, which
removes `deleteTempFile`, the inner try, the `renamed` branch, and the three
rename/cleanup test cases with their FileSystem subclasses. The two remaining
injection filesystems would then match `_num_partitions` instead of the temp
name, and the write-failure case would assert that the reader rejects the
partial file. Not blocking.
--
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]
