cloud-fan commented on code in PR #58167:
URL: https://github.com/apache/spark/pull/58167#discussion_r4071058970
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -268,6 +289,81 @@ 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 is a 1-byte format version followed by a 4-byte
big-endian Int.
+ * 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 {
+ dos.writeByte(1) // format version
+ dos.writeInt(partitionCount)
+ } {
+ 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")
+ }
+ logDebug(s"Written partition count $partitionCount to $countFilePath")
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Error writing partition count to ${MDC(PATH,
checkpointDirPath)}, " +
+ log"truncation detection will be inactive for this directory", e)
+ }
+ }
+
+ /**
+ * Read the expected partition count from the checkpoint directory metadata
file, if present.
+ * Returns [[None]] when the file is absent (checkpoint written by an older
Spark version)
+ * or unreadable, so callers must tolerate a missing value. See SPARK-58883.
+ */
+ private def readPartitionCountFromCheckpointDir(
+ sc: SparkContext, checkpointDirPath: String): Option[Int] = {
+ try {
+ val bufferSize = sc.conf.get(BUFFER_SIZE)
+ val countFilePath = new Path(checkpointDirPath,
checkpointPartitionCountFileName())
+ val fs = countFilePath.getFileSystem(sc.hadoopConfiguration)
+ val fileInputStream = fs.open(countFilePath, bufferSize)
+ val count = Utils.tryWithSafeFinally {
+ val dis = new DataInputStream(fileInputStream)
+ val version = dis.readByte()
+ if (version != 1) {
+ throw new IllegalArgumentException(
+ s"Unsupported _num_partitions format version: $version in
$countFilePath")
+ }
+ dis.readInt()
Review Comment:
**Non-blocking (P2):** `readInt()` makes every version-1 five-byte payload
authoritative. If the count bytes of an intact four-part checkpoint are
corrupted to another valid Int, such as 5, this path throws
`CHECKPOINT_TRUNCATED_DIRECTORY` instead of taking the corrupt-metadata
fallback; the added corruption test covers only a short payload that fails
parsing. Please make the unreleased payload self-validating and reject
corrupted count bytes before comparing them, while preserving the real
trailing-file mismatch behavior.
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -268,6 +289,81 @@ 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 is a 1-byte format version followed by a 4-byte
big-endian Int.
+ * 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 {
+ dos.writeByte(1) // format version
+ dos.writeInt(partitionCount)
+ } {
+ dos.close()
+ }
+ if (!fs.rename(tmpFilePath, countFilePath)) {
Review Comment:
**Non-blocking (P2):** The best-effort publication contract is not
regression-tested here: all new cases use a filesystem where `_num_partitions`
creation and rename succeed. Please add failure-injection coverage for both a
metadata create/write exception and a false rename result, asserting that
checkpointing still completes and the inactive-check warning is emitted.
Otherwise either branch could later start failing an action after its partition
files were already committed without making this suite red.
--
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]
