diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index 1e8be8d30cf39..93fe36a7fc0cf 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -993,6 +993,13 @@ }, "sqlState" : "XX000" }, + "CHECKPOINT_DIRECTORY_NOT_SET" : { + "message" : [ + "Cannot checkpoint because no checkpoint directory is configured.", + "Set one with `SparkContext.setCheckpointDir` or the \"spark.checkpoint.dir\" configuration." + ], + "sqlState" : "55019" + }, "CHECKPOINT_FILE_CHECKSUM_VERIFICATION_FAILED" : { "message" : [ "Checksum verification failed, the file may be corrupted. File: ", @@ -1008,6 +1015,13 @@ ], "sqlState" : "56000" }, + "CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH" : { + "message" : [ + "The checkpoint of RDD has partition(s), but the RDD itself has . The checkpoint RDD is .", + "This usually means the checkpoint directory is not on storage that both the driver and the executors can read and write, or that some checkpoint files did not survive until the checkpoint was read back." + ], + "sqlState" : "58030" + }, "CHECK_CONSTRAINT_VIOLATION" : { "message" : [ "CHECK constraint violated by row with values:", @@ -2662,6 +2676,12 @@ ], "sqlState" : "42822" }, + "FAILED_CREATE_CHECKPOINT_DIRECTORY" : { + "message" : [ + "Failed to create the checkpoint directory as FileSystem.mkdirs returned false." + ], + "sqlState" : "58030" + }, "FAILED_EXECUTE_UDF" : { "message" : [ "User defined function (: () => ) failed due to: ." @@ -3758,6 +3778,12 @@ }, "sqlState" : "42K03" }, + "INVALID_CHECKPOINT_DIRECTORY" : { + "message" : [ + "Cannot read the checkpoint directory : expected the partition file but found . The partition files must be numbered contiguously from part-00000." + ], + "sqlState" : "58030" + }, "INVALID_CLONE_SESSION_REQUEST" : { "message" : [ "Invalid session clone request." @@ -11528,33 +11554,6 @@ "countByValueApprox() does not support arrays" ] }, - "_LEGACY_ERROR_TEMP_3016" : { - "message" : [ - "Checkpoint directory has not been set in the SparkContext" - ] - }, - "_LEGACY_ERROR_TEMP_3017" : { - "message" : [ - "Invalid checkpoint file: " - ] - }, - "_LEGACY_ERROR_TEMP_3018" : { - "message" : [ - "Failed to create checkpoint path " - ] - }, - "_LEGACY_ERROR_TEMP_3019" : { - "message" : [ - "Checkpoint RDD has a different number of partitions from original RDD. Original", - "RDD [ID: , num of partitions: ];", - "Checkpoint RDD [ID: , num of partitions: ]." - ] - }, - "_LEGACY_ERROR_TEMP_3020" : { - "message" : [ - "Checkpoint dir must be specified." - ] - }, "_LEGACY_ERROR_TEMP_3028" : { "message" : [ "" diff --git a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala index cf931707c89e7..442d4623c4a92 100644 --- a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala +++ b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala @@ -157,22 +157,30 @@ private[spark] object SparkCoreErrors { def checkpointDirectoryHasNotBeenSetInSparkContextError(): Throwable = { new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3016", messageParameters = Map.empty, cause = null + errorClass = "CHECKPOINT_DIRECTORY_NOT_SET", + messageParameters = Map.empty, + cause = null ) } - def invalidCheckpointFileError(path: Path): Throwable = { + def invalidCheckpointDirectoryError( + partitionFilePath: Path, + expectedFileName: String): Throwable = { new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3017", - messageParameters = Map("path" -> s"$path"), + errorClass = "INVALID_CHECKPOINT_DIRECTORY", + messageParameters = Map( + "path" -> s"${partitionFilePath.getParent}", + "expectedFileName" -> expectedFileName, + "fileName" -> partitionFilePath.getName + ), cause = null ) } def failToCreateCheckpointPathError(checkpointDirPath: Path): Throwable = { new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3018", - messageParameters = Map("checkpointDirPath" -> s"$checkpointDirPath"), + errorClass = "FAILED_CREATE_CHECKPOINT_DIRECTORY", + messageParameters = Map("path" -> s"$checkpointDirPath"), cause = null ) } @@ -183,7 +191,7 @@ private[spark] object SparkCoreErrors { newRDDId: Int, newRDDLength: Int): Throwable = { new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3019", + errorClass = "CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH", messageParameters = Map( "originalRDDId" -> s"$originalRDDId", "originalRDDLength" -> s"$originalRDDLength", @@ -200,9 +208,8 @@ private[spark] object SparkCoreErrors { } def mustSpecifyCheckpointDirError(): Throwable = { - new SparkException( - errorClass = "_LEGACY_ERROR_TEMP_3020", messageParameters = Map.empty, cause = null - ) + SparkException.internalError( + "SparkContext.checkpointDir is unset when creating ReliableRDDCheckpointData.") } def askStandaloneSchedulerToShutDownExecutorsError(e: Exception): Throwable = { diff --git a/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala b/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala index fd42cea795d60..156a94abd6281 100644 --- a/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala +++ b/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala @@ -78,8 +78,9 @@ private[spark] class ReliableCheckpointRDD[T: ClassTag]( .sortBy(_.getName.stripPrefix("part-").toInt) // Fail fast if input files are invalid inputFiles.zipWithIndex.foreach { case (path, i) => - if (path.getName != ReliableCheckpointRDD.checkpointFileName(i)) { - throw SparkCoreErrors.invalidCheckpointFileError(path) + val expectedFileName = ReliableCheckpointRDD.checkpointFileName(i) + if (path.getName != expectedFileName) { + throw SparkCoreErrors.invalidCheckpointDirectoryError(path, expectedFileName) } } Array.tabulate(inputFiles.length)(i => new CheckpointRDDPartition(i)) diff --git a/core/src/test/scala/org/apache/spark/CheckpointSuite.scala b/core/src/test/scala/org/apache/spark/CheckpointSuite.scala index 58512a2282ac2..976e9b94d23cd 100644 --- a/core/src/test/scala/org/apache/spark/CheckpointSuite.scala +++ b/core/src/test/scala/org/apache/spark/CheckpointSuite.scala @@ -21,7 +21,7 @@ import java.io.File import scala.reflect.ClassTag -import org.apache.hadoop.fs.Path +import org.apache.hadoop.fs.{LocalFileSystem, Path} import org.apache.spark.internal.config.CACHE_CHECKPOINT_PREFERRED_LOCS_EXPIRE_TIME import org.apache.spark.internal.config.UI._ @@ -684,4 +684,71 @@ class CheckpointStorageSuite extends SparkFunSuite with LocalSparkContext { assert(flatMappedRDD.collect() === result) } } + + test("checkpoint() without a checkpoint directory") { + sc = new SparkContext("local", "test", new SparkConf().set(UI_ENABLED.key, "false")) + checkError( + exception = intercept[SparkException](sc.makeRDD(1 to 4).checkpoint()), + condition = "CHECKPOINT_DIRECTORY_NOT_SET", + sqlState = Some("55019")) + } + + test("reading a checkpoint directory with a missing partition file") { + withTempDir { checkpointDir => + sc = new SparkContext("local", "test", new SparkConf().set(UI_ENABLED.key, "false")) + sc.setCheckpointDir(checkpointDir.toString) + val rdd = sc.makeRDD(1 to 20, numSlices = 4) + rdd.checkpoint() + assert(rdd.collect().toSeq === (1 to 20)) + + // Drop a middle partition file so the remaining names are no longer part-00000..part-0000N. + val checkpointPath = new Path(rdd.getCheckpointFile.get) + val fs = checkpointPath.getFileSystem(sc.hadoopConfiguration) + assert(fs.delete(new Path(checkpointPath, "part-00001"), false)) + + val recovered = sc.checkpointFile[Int](checkpointPath.toString) + checkError( + exception = intercept[SparkException](recovered.partitions), + condition = "INVALID_CHECKPOINT_DIRECTORY", + sqlState = Some("58030"), + parameters = Map( + "path" -> checkpointPath.toString, + "expectedFileName" -> "part-00001", + "fileName" -> "part-00002")) + } + } + + test("checkpoint path that cannot be created") { + withTempDir { checkpointDir => + // MkdirsFailingFilesystem refuses to create the per-RDD directory and reports it the way + // HDFS and S3A do, by returning false rather than throwing. + val conf = new SparkConf() + .set("spark.hadoop.fs.file.impl", classOf[MkdirsFailingFilesystem].getName) + .set("spark.hadoop.fs.file.impl.disable.cache", "true") + .set(UI_ENABLED.key, "false") + sc = new SparkContext("local", "test", conf) + sc.setCheckpointDir(checkpointDir.toString) + val rdd = sc.makeRDD(1 to 4) + val rddPath = ReliableRDDCheckpointData.checkpointPath(sc, rdd.id).get + + rdd.checkpoint() + checkError( + exception = intercept[SparkException](rdd.collect()), + condition = "FAILED_CREATE_CHECKPOINT_DIRECTORY", + sqlState = Some("58030"), + parameters = Map("path" -> rddPath.toString)) + } + } +} + +/** + * A filesystem that refuses to create a per-RDD checkpoint directory, returning false from + * `mkdirs` instead of throwing. `LocalFileSystem` throws `FileAlreadyExistsException` when the + * path is taken, while HDFS and S3A can report the failure through the return value, which is + * what `ReliableCheckpointRDD` checks. + */ +class MkdirsFailingFilesystem extends LocalFileSystem { + override def mkdirs(f: Path): Boolean = { + if (f.getName.startsWith("rdd-")) false else super.mkdirs(f) + } }