You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "micheal-o (via GitHub)" <gi...@apache.org> on 2025/05/01 06:18:14 UTC
[PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
micheal-o opened a new pull request, #50773:
URL: https://github.com/apache/spark/pull/50773
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Introducing file integrity verification for state store files by also generating and uploading checksum file. This allows us to verify the file checksum during read. Introduced a new spark conf to enable/disable this (enabled by default). This can be enabled then disabled and vice versa, on the same checkpoint location, giving users the flexibility to turn on/off as needed.
It is used for the following state files: delta, snapshot, changelog, zip
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Integrity verification to detect file corruption and provide necessary additional information.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Added additional tests
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072018624
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1459,6 +1472,193 @@ abstract class StateStoreSuiteBase[ProviderClass <: StateStoreProvider]
}
}
+ testWithAllCodec("file checksum can be enabled and disabled for the same checkpoint") {
+ _ =>
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ var version = 0L
+
+ // Commit to store using file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val store = provider.getStore(version)
+ put(store, "1", 11, 100)
+ put(store, "2", 22, 200)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit without file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 1)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+
+ put(store, "3", 33, 300)
+ put(store, "4", 44, 400)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit with file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 2)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+ assert(get(store, "3", 33) === Some(300))
+ assert(get(store, "4", 44) === Some(400))
+
+ put(store, "5", 55, 500)
+ version = store.commit()
+ }
+ }
+ }
+
+ test("checksum files are also cleaned up during maintenance") {
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ val numBatches = 6
+ val minDeltas = 2
+ // Adding 1 to ensure snapshot is uploaded.
+ // Snapshot upload might happen at minDeltas or minDeltas + 1, depending on the provider
+ val maintFrequency = minDeltas + 1
+ var version = 0L
+
+ withSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> minDeltas.toString,
+ SQLConf.MIN_BATCHES_TO_RETAIN.key -> "1",
+ // So that RocksDB will also generate changelog files
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" ->
+ true.toString) {
+
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (version + 1 to numBatches).foreach { i =>
+ version = putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = i % maintFrequency == 0)
+ }
+
+ // This is because, hdfs and rocksdb old files detection logic is different
+ provider match {
+ case _: HDFSBackedStateStoreProvider =>
+ // For HDFS State store, files left:
+ // 3.delta to 6.delta (+ checksum file)
+ // 3.snapshot (+ checksum file), 6.snapshot (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 12, expectedNumChecksumFiles = 6)
+ case _ =>
+ // For RocksDB State store, files left:
+ // 6.changelog (+ checksum file), 6.zip (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 4, expectedNumChecksumFiles = 2)
+ }
+ }
+
+ // turn off file checksum, and verify that the previously created checksum files
+ // will be deleted by maintenance
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (version + 1 to version + numBatches).foreach { i =>
+ version = putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = i % maintFrequency == 0)
+ }
+
+ // now verify no checksum files are left
+ // This is because, hdfs and rocksdb old files detection logic is different
+ provider match {
+ case _: HDFSBackedStateStoreProvider =>
+ // For HDFS State store, files left:
+ // 6.delta, 9.delta to 12.delta
+ // 9.snapshot, 12.snapshot
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 7, expectedNumChecksumFiles = 0)
+ case _ =>
+ // For RocksDB State store, files left:
+ // 12.changelog, 12.zip
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 2, expectedNumChecksumFiles = 0)
+ }
+ }
+ }
+ }
+ }
+
+ testWithAllCodec("overwrite state file without overwriting checksum file") { _ =>
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ val numBatches = 3
+ val minDeltas = 2
+
+ withSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> minDeltas.toString,
+ // So that RocksDB will also generate changelog files
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" ->
+ true.toString) {
+
+ // First run with file checksum enabled. It will generate state and checksum files.
+ // Turn off file checksum, and regenerate only the state files
+ Seq(true, false).foreach { fileChecksumEnabled =>
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> fileChecksumEnabled.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (1 to numBatches).foreach { i =>
+ putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = false)
+ }
+
+ // This should only create snapshot and no delete
+ provider.doMaintenance()
+
+ // number of files should be the same.
+ // 3 delta/changelog files, 1 snapshot (with checksum files)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 8, expectedNumChecksumFiles = 4)
+ }
+ }
+ }
+
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ // now try to load the store with checksum enabled.
+ // It will verify the overwritten state files with the checksum files.
+ (1 to numBatches).foreach { i =>
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ // load from DFS should be successful
+ val store = provider.getStore(i)
+ store.abort()
+ }
+ }
+ }
+ }
+ }
+
+ private def verifyChecksumFiles(
Review Comment:
Are we covering all the files we are adding the checksum for ? i.e. do we have the necessary test coverage for all the types of files now ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072000866
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,513 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used and be easily readable.
+ // We can change this to byte array later if we start using algos with large values.
+ value: String,
+ mainFileSize: Long,
+ timestampMs: Long,
+ creator: ChecksumFileCreatorInfo) {
+
+ import Checksum._
+
+ def json(): String = {
+ mapper.writeValueAsString(this)
+ }
+}
+
+object Checksum {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ /** Used to convert between class and JSON. */
+ lazy val mapper = {
+ val _mapper = new ObjectMapper with ClassTagExtensions
+ _mapper.setSerializationInclusion(Include.NON_ABSENT)
+ _mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ _mapper.registerModule(DefaultScalaModule)
+ _mapper
+ }
+
+ def fromJson(json: String): Checksum = {
+ Serialization.read[Checksum](json)
+ }
+}
+
+/** Holds the path of the checksum file and knows the main file path. */
+case class ChecksumFile(path: Path) {
+ import ChecksumCheckpointFileManager._
+ assert(isChecksumFile(path), "path is not a checksum file")
+
+ val mainFilePath = new Path(path.toString.stripSuffix(CHECKSUM_FILE_SUFFIX))
+
+ /** The name of the file without any extensions e.g. my-file.txt.crc returns my-file */
+ val baseName: String = path.getName.split("\\.").head
+}
+
+/**
+ * A [[CheckpointFileManager]] that creates a checksum file for the main file.
+ * This wraps another [[CheckpointFileManager]] and adds checksum functionality on top of it.
+ * Under the hood, when a file is created, it also creates a checksum file with the same name as
+ * the main file but adds a suffix. It returns [[ChecksumCancellableFSDataOutputStream]]
+ * which handles the writing of the main file and checksum file.
+ *
+ * When a file is opened, it returns [[ChecksumFSDataInputStream]], which handles reading
+ * the main file and checksum file and does the checksum verification.
+ *
+ * In order to reduce the impact of reading/writing 2 files instead of 1, it uses a threadpool
+ * to read/write both files concurrently.
+ *
+ * @note
+ * It is able to read files written by other [[CheckpointFileManager]], that don't have checksum.
+ * It automatically deletes the checksum file when the main file is deleted.
+ * If you delete the main file with a different type of manager, then the checksum file will be
+ * left behind (i.e. orphan checksum file), since they don't know about it. It would be your
+ * responsibility to delete the orphan checksum files.
+ *
+ * @param underlyingFileMgr The file manager to use under the hood
+ * @param allowConcurrentDelete If true, allows deleting the main and checksum file concurrently.
+ * This is a perf optimization, but can potentially lead to
+ * orphan checksum files. If using this, it is your responsibility
+ * to clean up the potential orphan checksum files.
+ * @param numThreads This is the number of threads to use for the thread pool, for reading/writing
+ * files. To avoid blocking, if the file manager instance is being used by a
+ * single thread, then you can set this to 2 (one thread for main file, another
+ * for checksum file).
+ * If file manager is shared by multiple threads, you can set it to
+ * number of threads using file manager * 2.
+ * Setting this differently can lead to file operation being blocked waiting for
+ * a free thread.
+ */
+class ChecksumCheckpointFileManager(
+ private val underlyingFileMgr: CheckpointFileManager,
+ val allowConcurrentDelete: Boolean = false,
+ val numThreads: Int)
+ extends CheckpointFileManager with Logging {
+ assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
+ "and another for the checksum file")
+
+ import ChecksumCheckpointFileManager._
+
+ // This allows us to concurrently read/write the main file and checksum file
+ private val threadPool = ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+
+ override def list(path: Path, filter: PathFilter): Array[FileStatus] = {
+ underlyingFileMgr.list(path, filter)
+ }
+
+ override def mkdirs(path: Path): Unit = {
+ underlyingFileMgr.mkdirs(path)
+ }
+
+ override def createAtomic(path: Path,
+ overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
+ assert(!isChecksumFile(path), "Cannot directly create a checksum file")
+
+ val mainFileFuture = Future {
+ underlyingFileMgr.createAtomic(path, overwriteIfPossible)
+ }(threadPool)
+
+ val checksumFileFuture = Future {
+ underlyingFileMgr.createAtomic(getChecksumPath(path), overwriteIfPossible)
+ }(threadPool)
+
+ new ChecksumCancellableFSDataOutputStream(
+ awaitResult(mainFileFuture, Duration.Inf),
+ path,
+ awaitResult(checksumFileFuture, Duration.Inf),
+ threadPool
+ )
+ }
+
+ override def open(path: Path): FSDataInputStream = {
+ assert(!isChecksumFile(path), "Cannot directly open a checksum file")
+
+ val checksumInputStreamFuture = Future {
+ try {
+ Some(underlyingFileMgr.open(getChecksumPath(path)))
+ } catch {
+ // In case the client previously had file checksum disabled.
+ // Then previously created files won't have checksum.
+ case _: FileNotFoundException =>
+ logWarning(log"No checksum file found for ${MDC(PATH, path)}, " +
+ log"hence no checksum verification.")
+ None
+ }
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.open(path)
+ }(threadPool)
+
+ val mainStream = awaitResult(mainInputStreamFuture, Duration.Inf)
+ val checksumStream = awaitResult(checksumInputStreamFuture, Duration.Inf)
+
+ checksumStream.map { chkStream =>
+ new ChecksumFSDataInputStream(mainStream, path, chkStream, threadPool)
+ }.getOrElse(mainStream)
+ }
+
+ override def exists(path: Path): Boolean = underlyingFileMgr.exists(path)
+
+ override def delete(path: Path): Unit = {
+ // Allowing directly deleting the checksum file for orphan checksum file scenario
+ if (isChecksumFile(path)) {
+ deleteChecksumFile(path)
+ } else if (allowConcurrentDelete) {
+ // Ideally, we should first try to delete the checksum file
+ // before the main file, to avoid a situation where the main file is deleted but the
+ // checksum file deletion failed. The client might not call delete again if the main file
+ // no longer exists.
+ // But if allowConcurrentDelete is enabled, then we can do it concurrently for perf.
+ // But the client would be responsible for cleaning up potential orphan checksum files
+ // if it happens.
+ val checksumInputStreamFuture = Future {
+ deleteChecksumFile(getChecksumPath(path))
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.delete(path)
+ }(threadPool)
+
+ awaitResult(mainInputStreamFuture, Duration.Inf)
+ awaitResult(checksumInputStreamFuture, Duration.Inf)
+ } else {
+ // First delete the checksum file, then main file
+ deleteChecksumFile(getChecksumPath(path))
+ underlyingFileMgr.delete(path)
+ }
+ }
+
+ private def deleteChecksumFile(checksumPath: Path): Unit = {
+ try {
+ underlyingFileMgr.delete(checksumPath)
+ logDebug(log"Deleted checksum file ${MDC(PATH, checksumPath)}")
+ } catch {
+ case _: FileNotFoundException =>
+ // Ignore if file has already been deleted
+ // or the main file was created initially without checksum
+ logWarning(log"Skipping deletion of checksum file ${MDC(PATH, checksumPath)} " +
Review Comment:
Will this also apply to existing checkpoints also that dont have checksum verification enabled ? should we log only if the feature is enabled ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #50773: [SPARK-51972][SS] State Store file integrity verification using checksum
URL: https://github.com/apache/spark/pull/50773
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2070916110
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala:
##########
@@ -1247,6 +1255,8 @@ class RocksDB(
silentDeleteRecursively(localRootDir, "closing RocksDB")
// Clear internal maps to reset the state
clearColFamilyMaps()
+
Review Comment:
nit: intentional ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -342,6 +363,8 @@ class RocksDBFileManager(
} else {
// Delete all non-immutable files in local dir, and unzip new ones from DFS commit file
listRocksDBFiles(localDir)._2.foreach(_.delete())
+ // TODO: We are using fs here to read the file, checksum verification wouldn't happen
Review Comment:
Do we need a separate SPARK ticket here ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala:
##########
@@ -661,6 +661,55 @@ object StateStoreProvider extends Logging {
}
}
+ /**
+ * If file checksum is enabled, then when we delete store files, using fm.delete,
+ * then the corresponding checksum file would be deleted too. But there are situations
+ * where we can have orphan checksum files (i.e. the main file no longer exist),
+ * so we need to delete them too:
+ * 1. If the file was created when file checksum was enabled, but then file checksum
+ * was later disabled. In this case, fm.delete will not delete the checksum file.
+ * Since we use a specific fm (checksum file manager) when checksum is enabled.
+ * 2. We enable concurrent file deletion for the checksum file manager,
+ * for perf improvement, hence there is a slight chance deleting the main file was
+ * successful but the checksum file deletion failed. Hence, we need to clean it up by
+ * ourselves. If we don't want this behavior we can turn off concurrent deletion
+ * for the checksum file manager
+ * */
+ private[streaming] def deleteOrphanChecksumFiles(
+ fm: CheckpointFileManager,
+ checksumFiles: Seq[ChecksumFile],
+ deletedStoreFiles: Seq[Path],
+ minVersionToRetain: Long,
+ fileChecksumEnabled: Boolean): Unit = {
+ // Use file name instead since path format might be different
+ // i.e. checksum files might have scheme prefix and the deleted files might not.
+ val deletedStoreFilesSet = deletedStoreFiles.map(_.getName).toSet
+ val oldChecksumFiles = checksumFiles
+ .filter(f => f.baseName.split("_")(0).toLong < minVersionToRetain)
+
+ val orphanChecksumFiles = if (fileChecksumEnabled) {
+ oldChecksumFiles
+ // old checksum files and was not part of the store files deleted
+ .filterNot(f => deletedStoreFilesSet.contains(f.mainFilePath.getName))
+ } else {
+ // all old checksum files, in case file checksum was previously enabled
+ oldChecksumFiles
+ }
+
+ val (_, dur) = Utils.timeTakenMs {
+ orphanChecksumFiles.foreach { f =>
+ fm.delete(f.path)
Review Comment:
Is it possible for these operations to potentially fail with exception ? should we catch and log that here ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072005201
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala:
##########
@@ -661,6 +661,55 @@ object StateStoreProvider extends Logging {
}
}
+ /**
+ * If file checksum is enabled, then when we delete store files, using fm.delete,
+ * then the corresponding checksum file would be deleted too. But there are situations
+ * where we can have orphan checksum files (i.e. the main file no longer exist),
+ * so we need to delete them too:
+ * 1. If the file was created when file checksum was enabled, but then file checksum
+ * was later disabled. In this case, fm.delete will not delete the checksum file.
+ * Since we use a specific fm (checksum file manager) when checksum is enabled.
+ * 2. We enable concurrent file deletion for the checksum file manager,
+ * for perf improvement, hence there is a slight chance deleting the main file was
+ * successful but the checksum file deletion failed. Hence, we need to clean it up by
+ * ourselves. If we don't want this behavior we can turn off concurrent deletion
+ * for the checksum file manager
+ * */
+ private[streaming] def deleteOrphanChecksumFiles(
+ fm: CheckpointFileManager,
+ checksumFiles: Seq[ChecksumFile],
+ deletedStoreFiles: Seq[Path],
+ minVersionToRetain: Long,
+ fileChecksumEnabled: Boolean): Unit = {
+ // Use file name instead since path format might be different
+ // i.e. checksum files might have scheme prefix and the deleted files might not.
+ val deletedStoreFilesSet = deletedStoreFiles.map(_.getName).toSet
+ val oldChecksumFiles = checksumFiles
+ .filter(f => f.baseName.split("_")(0).toLong < minVersionToRetain)
+
+ val orphanChecksumFiles = if (fileChecksumEnabled) {
+ oldChecksumFiles
+ // old checksum files and was not part of the store files deleted
+ .filterNot(f => deletedStoreFilesSet.contains(f.mainFilePath.getName))
+ } else {
+ // all old checksum files, in case file checksum was previously enabled
+ oldChecksumFiles
+ }
+
+ val (_, dur) = Utils.timeTakenMs {
+ orphanChecksumFiles.foreach { f =>
+ fm.delete(f.path)
Review Comment:
done
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072014351
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala:
##########
@@ -1072,7 +1072,8 @@ class StateDataSourceTransformWithStateSuite extends StateStoreMetricsTest
val dfsRootDir = new File(tmpDir.getAbsolutePath + "/state/0/4")
val fileManager = new RocksDBFileManager(
dfsRootDir.getAbsolutePath, Utils.createTempDir(), new Configuration,
- CompressionCodec.LZ4)
+ CompressionCodec.LZ4, fileChecksumEnabled = SQLConf.get.checkpointFileChecksumEnabled,
Review Comment:
@micheal-o - do we need to add any additional tests to test state data source reader with integrity verification enabled ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072032193
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1459,6 +1472,193 @@ abstract class StateStoreSuiteBase[ProviderClass <: StateStoreProvider]
}
}
+ testWithAllCodec("file checksum can be enabled and disabled for the same checkpoint") {
+ _ =>
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ var version = 0L
+
+ // Commit to store using file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val store = provider.getStore(version)
+ put(store, "1", 11, 100)
+ put(store, "2", 22, 200)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit without file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 1)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+
+ put(store, "3", 33, 300)
+ put(store, "4", 44, 400)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit with file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 2)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+ assert(get(store, "3", 33) === Some(300))
+ assert(get(store, "4", 44) === Some(400))
+
+ put(store, "5", 55, 500)
+ version = store.commit()
+ }
+ }
+ }
+
+ test("checksum files are also cleaned up during maintenance") {
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ val numBatches = 6
+ val minDeltas = 2
+ // Adding 1 to ensure snapshot is uploaded.
+ // Snapshot upload might happen at minDeltas or minDeltas + 1, depending on the provider
+ val maintFrequency = minDeltas + 1
+ var version = 0L
+
+ withSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> minDeltas.toString,
+ SQLConf.MIN_BATCHES_TO_RETAIN.key -> "1",
+ // So that RocksDB will also generate changelog files
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" ->
+ true.toString) {
+
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (version + 1 to numBatches).foreach { i =>
+ version = putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = i % maintFrequency == 0)
+ }
+
+ // This is because, hdfs and rocksdb old files detection logic is different
+ provider match {
+ case _: HDFSBackedStateStoreProvider =>
+ // For HDFS State store, files left:
+ // 3.delta to 6.delta (+ checksum file)
+ // 3.snapshot (+ checksum file), 6.snapshot (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 12, expectedNumChecksumFiles = 6)
+ case _ =>
+ // For RocksDB State store, files left:
+ // 6.changelog (+ checksum file), 6.zip (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 4, expectedNumChecksumFiles = 2)
+ }
+ }
+
+ // turn off file checksum, and verify that the previously created checksum files
+ // will be deleted by maintenance
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (version + 1 to version + numBatches).foreach { i =>
+ version = putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = i % maintFrequency == 0)
+ }
+
+ // now verify no checksum files are left
+ // This is because, hdfs and rocksdb old files detection logic is different
+ provider match {
+ case _: HDFSBackedStateStoreProvider =>
+ // For HDFS State store, files left:
+ // 6.delta, 9.delta to 12.delta
+ // 9.snapshot, 12.snapshot
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 7, expectedNumChecksumFiles = 0)
+ case _ =>
+ // For RocksDB State store, files left:
+ // 12.changelog, 12.zip
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 2, expectedNumChecksumFiles = 0)
+ }
+ }
+ }
+ }
+ }
+
+ testWithAllCodec("overwrite state file without overwriting checksum file") { _ =>
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ val numBatches = 3
+ val minDeltas = 2
+
+ withSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> minDeltas.toString,
+ // So that RocksDB will also generate changelog files
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" ->
+ true.toString) {
+
+ // First run with file checksum enabled. It will generate state and checksum files.
+ // Turn off file checksum, and regenerate only the state files
+ Seq(true, false).foreach { fileChecksumEnabled =>
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> fileChecksumEnabled.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (1 to numBatches).foreach { i =>
+ putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = false)
+ }
+
+ // This should only create snapshot and no delete
+ provider.doMaintenance()
+
+ // number of files should be the same.
+ // 3 delta/changelog files, 1 snapshot (with checksum files)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 8, expectedNumChecksumFiles = 4)
+ }
+ }
+ }
+
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ // now try to load the store with checksum enabled.
+ // It will verify the overwritten state files with the checksum files.
+ (1 to numBatches).foreach { i =>
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ // load from DFS should be successful
+ val store = provider.getStore(i)
+ store.abort()
+ }
+ }
+ }
+ }
+ }
+
+ private def verifyChecksumFiles(
Review Comment:
yeah this covers all the file types we are enabling for. This runs for HDFS and RocksDB state store with all the different compression types. For RocksDB also runs for avro too.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2070974574
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,512 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used
+ value: String,
Review Comment:
This is the checksum value right ? why not just use a byte array here ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2070914229
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3126,6 +3126,15 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED =
+ buildConf("spark.sql.streaming.checkpoint.fileChecksum.enabled")
+ .internal()
+ .doc("When true, checksum would be generated and verified for some checkpoint files. " +
Review Comment:
nit: lets just say `verified for checkpoint files` ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072027623
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceTransformWithStateSuite.scala:
##########
@@ -1072,7 +1072,8 @@ class StateDataSourceTransformWithStateSuite extends StateStoreMetricsTest
val dfsRootDir = new File(tmpDir.getAbsolutePath + "/state/0/4")
val fileManager = new RocksDBFileManager(
dfsRootDir.getAbsolutePath, Utils.createTempDir(), new Configuration,
- CompressionCodec.LZ4)
+ CompressionCodec.LZ4, fileChecksumEnabled = SQLConf.get.checkpointFileChecksumEnabled,
Review Comment:
No need for additional tests specific to state reader. Existing reader tests are exercising that already.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #50773:
URL: https://github.com/apache/spark/pull/50773#issuecomment-3193061842
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072009335
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,513 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used and be easily readable.
+ // We can change this to byte array later if we start using algos with large values.
+ value: String,
+ mainFileSize: Long,
+ timestampMs: Long,
+ creator: ChecksumFileCreatorInfo) {
+
+ import Checksum._
+
+ def json(): String = {
+ mapper.writeValueAsString(this)
+ }
+}
+
+object Checksum {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ /** Used to convert between class and JSON. */
+ lazy val mapper = {
+ val _mapper = new ObjectMapper with ClassTagExtensions
+ _mapper.setSerializationInclusion(Include.NON_ABSENT)
+ _mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ _mapper.registerModule(DefaultScalaModule)
+ _mapper
+ }
+
+ def fromJson(json: String): Checksum = {
+ Serialization.read[Checksum](json)
+ }
+}
+
+/** Holds the path of the checksum file and knows the main file path. */
+case class ChecksumFile(path: Path) {
+ import ChecksumCheckpointFileManager._
+ assert(isChecksumFile(path), "path is not a checksum file")
+
+ val mainFilePath = new Path(path.toString.stripSuffix(CHECKSUM_FILE_SUFFIX))
+
+ /** The name of the file without any extensions e.g. my-file.txt.crc returns my-file */
+ val baseName: String = path.getName.split("\\.").head
+}
+
+/**
+ * A [[CheckpointFileManager]] that creates a checksum file for the main file.
+ * This wraps another [[CheckpointFileManager]] and adds checksum functionality on top of it.
+ * Under the hood, when a file is created, it also creates a checksum file with the same name as
+ * the main file but adds a suffix. It returns [[ChecksumCancellableFSDataOutputStream]]
+ * which handles the writing of the main file and checksum file.
+ *
+ * When a file is opened, it returns [[ChecksumFSDataInputStream]], which handles reading
+ * the main file and checksum file and does the checksum verification.
+ *
+ * In order to reduce the impact of reading/writing 2 files instead of 1, it uses a threadpool
+ * to read/write both files concurrently.
+ *
+ * @note
+ * It is able to read files written by other [[CheckpointFileManager]], that don't have checksum.
+ * It automatically deletes the checksum file when the main file is deleted.
+ * If you delete the main file with a different type of manager, then the checksum file will be
+ * left behind (i.e. orphan checksum file), since they don't know about it. It would be your
+ * responsibility to delete the orphan checksum files.
+ *
+ * @param underlyingFileMgr The file manager to use under the hood
+ * @param allowConcurrentDelete If true, allows deleting the main and checksum file concurrently.
+ * This is a perf optimization, but can potentially lead to
+ * orphan checksum files. If using this, it is your responsibility
+ * to clean up the potential orphan checksum files.
+ * @param numThreads This is the number of threads to use for the thread pool, for reading/writing
+ * files. To avoid blocking, if the file manager instance is being used by a
+ * single thread, then you can set this to 2 (one thread for main file, another
+ * for checksum file).
+ * If file manager is shared by multiple threads, you can set it to
+ * number of threads using file manager * 2.
+ * Setting this differently can lead to file operation being blocked waiting for
+ * a free thread.
+ */
+class ChecksumCheckpointFileManager(
+ private val underlyingFileMgr: CheckpointFileManager,
+ val allowConcurrentDelete: Boolean = false,
+ val numThreads: Int)
+ extends CheckpointFileManager with Logging {
+ assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
+ "and another for the checksum file")
+
+ import ChecksumCheckpointFileManager._
+
+ // This allows us to concurrently read/write the main file and checksum file
+ private val threadPool = ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+
+ override def list(path: Path, filter: PathFilter): Array[FileStatus] = {
+ underlyingFileMgr.list(path, filter)
+ }
+
+ override def mkdirs(path: Path): Unit = {
+ underlyingFileMgr.mkdirs(path)
+ }
+
+ override def createAtomic(path: Path,
+ overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
+ assert(!isChecksumFile(path), "Cannot directly create a checksum file")
+
+ val mainFileFuture = Future {
+ underlyingFileMgr.createAtomic(path, overwriteIfPossible)
+ }(threadPool)
+
+ val checksumFileFuture = Future {
+ underlyingFileMgr.createAtomic(getChecksumPath(path), overwriteIfPossible)
+ }(threadPool)
+
+ new ChecksumCancellableFSDataOutputStream(
+ awaitResult(mainFileFuture, Duration.Inf),
+ path,
+ awaitResult(checksumFileFuture, Duration.Inf),
+ threadPool
+ )
+ }
+
+ override def open(path: Path): FSDataInputStream = {
+ assert(!isChecksumFile(path), "Cannot directly open a checksum file")
+
+ val checksumInputStreamFuture = Future {
+ try {
+ Some(underlyingFileMgr.open(getChecksumPath(path)))
+ } catch {
+ // In case the client previously had file checksum disabled.
+ // Then previously created files won't have checksum.
+ case _: FileNotFoundException =>
+ logWarning(log"No checksum file found for ${MDC(PATH, path)}, " +
+ log"hence no checksum verification.")
+ None
+ }
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.open(path)
+ }(threadPool)
+
+ val mainStream = awaitResult(mainInputStreamFuture, Duration.Inf)
+ val checksumStream = awaitResult(checksumInputStreamFuture, Duration.Inf)
+
+ checksumStream.map { chkStream =>
+ new ChecksumFSDataInputStream(mainStream, path, chkStream, threadPool)
+ }.getOrElse(mainStream)
+ }
+
+ override def exists(path: Path): Boolean = underlyingFileMgr.exists(path)
+
+ override def delete(path: Path): Unit = {
+ // Allowing directly deleting the checksum file for orphan checksum file scenario
+ if (isChecksumFile(path)) {
+ deleteChecksumFile(path)
+ } else if (allowConcurrentDelete) {
+ // Ideally, we should first try to delete the checksum file
+ // before the main file, to avoid a situation where the main file is deleted but the
+ // checksum file deletion failed. The client might not call delete again if the main file
+ // no longer exists.
+ // But if allowConcurrentDelete is enabled, then we can do it concurrently for perf.
+ // But the client would be responsible for cleaning up potential orphan checksum files
+ // if it happens.
+ val checksumInputStreamFuture = Future {
+ deleteChecksumFile(getChecksumPath(path))
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.delete(path)
+ }(threadPool)
+
+ awaitResult(mainInputStreamFuture, Duration.Inf)
+ awaitResult(checksumInputStreamFuture, Duration.Inf)
+ } else {
+ // First delete the checksum file, then main file
+ deleteChecksumFile(getChecksumPath(path))
+ underlyingFileMgr.delete(path)
+ }
+ }
+
+ private def deleteChecksumFile(checksumPath: Path): Unit = {
+ try {
+ underlyingFileMgr.delete(checksumPath)
+ logDebug(log"Deleted checksum file ${MDC(PATH, checksumPath)}")
+ } catch {
+ case _: FileNotFoundException =>
+ // Ignore if file has already been deleted
+ // or the main file was created initially without checksum
+ logWarning(log"Skipping deletion of checksum file ${MDC(PATH, checksumPath)} " +
+ log"since it does not exist.")
+ }
+ }
+
+ override def isLocal: Boolean = underlyingFileMgr.isLocal
+
+ override def createCheckpointDirectory(): Path = {
+ underlyingFileMgr.createCheckpointDirectory()
+ }
+
+ override def close(): Unit = {
+ threadPool.shutdown()
+ // Wait a bit for it to finish up in case there is any ongoing work
+ // Can consider making this timeout configurable, if needed
+ val timeoutMs = 100
+ if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
+ logWarning(log"Thread pool did not shutdown after ${MDC(TIMEOUT, timeoutMs)} ms")
+ }
+ }
+}
+
+private[streaming] object ChecksumCheckpointFileManager {
+ val CHECKSUM_FILE_SUFFIX = ".crc"
+
+ def awaitResult[T](future: Future[T], atMost: Duration): T = {
+ try {
+ ThreadUtils.awaitResult(future, atMost)
+ } catch {
+ // awaitResult wraps the exception. Unwrap it, and throw the actual error
+ case e: SparkException if e.getMessage.contains("Exception thrown in awaitResult") =>
+ throw e.getCause
+ }
+ }
+
+ def getChecksumPath(mainFilePath: Path): Path = {
+ new Path(mainFilePath.toString + CHECKSUM_FILE_SUFFIX)
+ }
+
+ def isChecksumFile(path: Path): Boolean = {
+ path.getName.endsWith(CHECKSUM_FILE_SUFFIX)
+ }
+}
+
+/** An implementation of [[FSDataInputStream]] that calculates the checksum of the file
+ * that the client is reading (main file) incrementally, while it is being read.
+ * It then does checksum verification on close, to verify that the computed checksum
+ * matches the expected checksum in the checksum file.
+ *
+ * Computing the checksum incrementally and doing the verification after file read is complete
+ * is for better performance, instead of first reading the entire file and doing verification
+ * before the client starts reading the file.
+ *
+ * @param mainStream Input stream for the main file the client wants to read
+ * @param path The path of the main file
+ * @param expectedChecksumStream The input stream for the checksum file
+ * @param threadPool Thread pool to use for concurrently operating on the main and checksum file
+ * */
+class ChecksumFSDataInputStream(
+ private val mainStream: FSDataInputStream,
+ path: Path,
+ private val expectedChecksumStream: FSDataInputStream,
+ private val threadPool: ExecutionContext)
+ extends FSDataInputStream(new CheckedSequentialInputStream(mainStream)) with Logging {
+
+ import ChecksumCheckpointFileManager._
+
+ @volatile private var verified = false
+ @volatile private var closed = false
+
+ override def close(): Unit = {
+ if (!closed) {
+ // We verify the checksum only when the client is done reading.
+ verifyChecksum()
+ closeInternal()
+ }
+ }
+
+ /** This is used to skip checksum verification on close.
+ * Avoid using this, and it is only used for a situation where the file is opened, read,
+ * then closed multiple times, and we want to avoid doing verification each time
+ * and only want to do it once.
+ * */
+ def closeWithoutChecksumVerification(): Unit = {
+ if (!closed) {
+ // Ideally this should be warning, but if a file is doing this frequently
+ // it will cause unnecessary noise in the logs. This can be changed later.
+ logDebug(log"Closing file ${MDC(PATH, path)} without checksum verification")
+ closeInternal()
+ }
+ }
+
+ private def closeInternal(): Unit = {
+ closed = true
+
+ val mainCloseFuture = Future {
+ super.close() // close the main file
+ }(threadPool)
+
+ val checksumCloseFuture = Future {
+ expectedChecksumStream.close() // close the checksum file
+ }(threadPool)
+
+ awaitResult(mainCloseFuture, Duration.Inf)
+ awaitResult(checksumCloseFuture, Duration.Inf)
+ }
+
+ private def verifyChecksum(): Unit = {
+ if (!verified) {
+ // It is possible the file was not read till the end by the reader,
+ // but we need the entire file content for checksum verification.
+ // Hence, we will read the file till the end from where reader stopped.
+ var remainingBytesRead = 0
+ val buffer = new Array[Byte](1024)
+ var bytesRead = 0
+ while ({ bytesRead = super.read(buffer); bytesRead != -1 }) {
+ remainingBytesRead += bytesRead
+ }
+
+ // we are at the end position so that tells us the size
+ val computedFileSize = mainStream.getPos
+
+ if (remainingBytesRead > 0) {
+ // Making this debug log since most of our files are not read exactly to the end
+ // and don't want to cause unnecessary noise in the logs.
+ logDebug(log"File ${MDC(PATH, path)} was not read till the end by reader. " +
+ log"Finished reading the rest of the file for checksum verification. " +
+ log"Remaining bytes read: ${MDC(NUM_BYTES, remainingBytesRead)}, " +
+ log"total size: ${MDC(NUM_BYTES, computedFileSize)}.")
+ }
+
+ // Read the expected checksum from the checksum file
+ val expectedChecksumJson = Source.fromInputStream(
+ expectedChecksumStream, StandardCharsets.UTF_8.name()).mkString
+ val expectedChecksum = Checksum.fromJson(expectedChecksumJson)
+ // Get what we computed while the main file was being read locally
+ // `in` is the CheckedSequentialInputStream we created
+ val computedChecksumValue = in.asInstanceOf[CheckedInputStream].getChecksum.getValue.toInt
+
+ logInfo(log"Verifying checksum for file ${MDC(PATH, path)}, " +
+ log"remainingBytesRead= ${MDC(NUM_BYTES, remainingBytesRead)}. " +
+ log"Computed(checksum= ${MDC(CHECKSUM, computedChecksumValue)}, " +
+ log"fileSize= ${MDC(NUM_BYTES, computedFileSize)})." +
+ log"Checksum file content: ${MDC(CHECKSUM, expectedChecksumJson)}")
+
+ verified = true
+
+ // Compare file size too, in case of collision
+ if (expectedChecksum.value.toInt != computedChecksumValue ||
+ expectedChecksum.mainFileSize != computedFileSize) {
+ throw QueryExecutionErrors.checkpointFileChecksumVerificationFailed(
+ path,
+ expectedSize = expectedChecksum.mainFileSize,
+ expectedChecksum.value,
+ computedSize = computedFileSize,
+ computedChecksumValue.toString)
+ }
+ }
+ }
+}
+
+/** This implements [[CheckedInputStream]] that allows us to compute the checksum as the file
+ * is being read. [[FSDataInputStream]] needs the passed in input stream to implement Seekable
+ * and PositionedReadable. We want the file to be read sequentially only to allow us to correctly
+ * compute the checksum. This is blocking the seekable apis from being used in the underlying stream
+ * */
+private class CheckedSequentialInputStream(data: InputStream)
+ // TODO: Make the checksum algo configurable
+ extends CheckedInputStream(data, new CRC32C())
+ with Seekable
+ with PositionedReadable {
+
+ // Seekable methods
+ override def seek(pos: Long): Unit = {
+ throw new UnsupportedOperationException("Seek not supported")
+ }
+
+ override def getPos: Long = {
+ throw new UnsupportedOperationException("getPos not supported")
+ }
+
+ override def seekToNewSource(targetPos: Long): Boolean = {
+ throw new UnsupportedOperationException("seekToNewSource not supported")
+ }
+
+ // PositionedReadable methods
+ def read(position: Long, buffer: Array[Byte], offset: Int, length: Int): Int = {
+ throw new UnsupportedOperationException("read from position not supported")
+ }
+
+ def readFully(position: Long, buffer: Array[Byte], offset: Int, length: Int): Unit = {
+ throw new UnsupportedOperationException("readFully from position not supported")
+ }
+
+ def readFully(position: Long, buffer: Array[Byte]): Unit = {
+ throw new UnsupportedOperationException("readFully from position not supported")
+ }
+}
+
+/** An implementation of [[CancellableFSDataOutputStream]] that calculates the checksum of the file
+ * that the client is writing (main file) incrementally, while it is being written.
+ * It then writes the main file and an additional checksum file, which will be used for verification
+ * by [[ChecksumFSDataInputStream]] on file read.
+ *
+ * @param mainStream Output stream for the main file the client wants to write to
+ * @param path The path of the main file
+ * @param checksumStream Output stream for the checksum file to write the computed checksum
+ * @param uploadThreadPool Thread pool used to concurrently upload the main and checksum file
+ * */
+class ChecksumCancellableFSDataOutputStream(
+ private val mainStream: CancellableFSDataOutputStream,
+ path: Path,
+ private val checksumStream: CancellableFSDataOutputStream,
+ private val uploadThreadPool: ExecutionContext)
+ // TODO: make the checksum algo configurable
Review Comment:
same here
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072005461
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -342,6 +363,8 @@ class RocksDBFileManager(
} else {
// Delete all non-immutable files in local dir, and unzip new ones from DFS commit file
listRocksDBFiles(localDir)._2.foreach(_.delete())
+ // TODO: We are using fs here to read the file, checksum verification wouldn't happen
Review Comment:
done: https://issues.apache.org/jira/browse/SPARK-51988
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072030700
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1459,6 +1472,193 @@ abstract class StateStoreSuiteBase[ProviderClass <: StateStoreProvider]
}
}
+ testWithAllCodec("file checksum can be enabled and disabled for the same checkpoint") {
+ _ =>
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ var version = 0L
+
+ // Commit to store using file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val store = provider.getStore(version)
+ put(store, "1", 11, 100)
+ put(store, "2", 22, 200)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit without file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 1)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+
+ put(store, "3", 33, 300)
+ put(store, "4", 44, 400)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit with file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 2)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+ assert(get(store, "3", 33) === Some(300))
+ assert(get(store, "4", 44) === Some(400))
+
+ put(store, "5", 55, 500)
+ version = store.commit()
+ }
+ }
+ }
+
+ test("checksum files are also cleaned up during maintenance") {
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ val numBatches = 6
+ val minDeltas = 2
+ // Adding 1 to ensure snapshot is uploaded.
+ // Snapshot upload might happen at minDeltas or minDeltas + 1, depending on the provider
+ val maintFrequency = minDeltas + 1
+ var version = 0L
+
+ withSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> minDeltas.toString,
+ SQLConf.MIN_BATCHES_TO_RETAIN.key -> "1",
+ // So that RocksDB will also generate changelog files
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" ->
+ true.toString) {
+
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (version + 1 to numBatches).foreach { i =>
+ version = putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = i % maintFrequency == 0)
+ }
+
+ // This is because, hdfs and rocksdb old files detection logic is different
+ provider match {
+ case _: HDFSBackedStateStoreProvider =>
+ // For HDFS State store, files left:
+ // 3.delta to 6.delta (+ checksum file)
+ // 3.snapshot (+ checksum file), 6.snapshot (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 12, expectedNumChecksumFiles = 6)
+ case _ =>
+ // For RocksDB State store, files left:
+ // 6.changelog (+ checksum file), 6.zip (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 4, expectedNumChecksumFiles = 2)
+ }
+ }
+
+ // turn off file checksum, and verify that the previously created checksum files
+ // will be deleted by maintenance
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
Review Comment:
Not sure I get what you mean here? I already have a test that verifies that we can read checkpoint written without checksum enabled.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072006492
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,512 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used
+ value: String,
Review Comment:
now using the int, we can use byte array later when we have other algos, in order to be agnostic. Added comment for that
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072017283
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,513 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used and be easily readable.
+ // We can change this to byte array later if we start using algos with large values.
+ value: String,
+ mainFileSize: Long,
+ timestampMs: Long,
+ creator: ChecksumFileCreatorInfo) {
+
+ import Checksum._
+
+ def json(): String = {
+ mapper.writeValueAsString(this)
+ }
+}
+
+object Checksum {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ /** Used to convert between class and JSON. */
+ lazy val mapper = {
+ val _mapper = new ObjectMapper with ClassTagExtensions
+ _mapper.setSerializationInclusion(Include.NON_ABSENT)
+ _mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ _mapper.registerModule(DefaultScalaModule)
+ _mapper
+ }
+
+ def fromJson(json: String): Checksum = {
+ Serialization.read[Checksum](json)
+ }
+}
+
+/** Holds the path of the checksum file and knows the main file path. */
+case class ChecksumFile(path: Path) {
+ import ChecksumCheckpointFileManager._
+ assert(isChecksumFile(path), "path is not a checksum file")
+
+ val mainFilePath = new Path(path.toString.stripSuffix(CHECKSUM_FILE_SUFFIX))
+
+ /** The name of the file without any extensions e.g. my-file.txt.crc returns my-file */
+ val baseName: String = path.getName.split("\\.").head
+}
+
+/**
+ * A [[CheckpointFileManager]] that creates a checksum file for the main file.
+ * This wraps another [[CheckpointFileManager]] and adds checksum functionality on top of it.
+ * Under the hood, when a file is created, it also creates a checksum file with the same name as
+ * the main file but adds a suffix. It returns [[ChecksumCancellableFSDataOutputStream]]
+ * which handles the writing of the main file and checksum file.
+ *
+ * When a file is opened, it returns [[ChecksumFSDataInputStream]], which handles reading
+ * the main file and checksum file and does the checksum verification.
+ *
+ * In order to reduce the impact of reading/writing 2 files instead of 1, it uses a threadpool
+ * to read/write both files concurrently.
+ *
+ * @note
+ * It is able to read files written by other [[CheckpointFileManager]], that don't have checksum.
+ * It automatically deletes the checksum file when the main file is deleted.
+ * If you delete the main file with a different type of manager, then the checksum file will be
+ * left behind (i.e. orphan checksum file), since they don't know about it. It would be your
+ * responsibility to delete the orphan checksum files.
+ *
+ * @param underlyingFileMgr The file manager to use under the hood
+ * @param allowConcurrentDelete If true, allows deleting the main and checksum file concurrently.
+ * This is a perf optimization, but can potentially lead to
+ * orphan checksum files. If using this, it is your responsibility
+ * to clean up the potential orphan checksum files.
+ * @param numThreads This is the number of threads to use for the thread pool, for reading/writing
+ * files. To avoid blocking, if the file manager instance is being used by a
+ * single thread, then you can set this to 2 (one thread for main file, another
+ * for checksum file).
+ * If file manager is shared by multiple threads, you can set it to
+ * number of threads using file manager * 2.
+ * Setting this differently can lead to file operation being blocked waiting for
+ * a free thread.
+ */
+class ChecksumCheckpointFileManager(
+ private val underlyingFileMgr: CheckpointFileManager,
+ val allowConcurrentDelete: Boolean = false,
+ val numThreads: Int)
+ extends CheckpointFileManager with Logging {
+ assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
+ "and another for the checksum file")
+
+ import ChecksumCheckpointFileManager._
+
+ // This allows us to concurrently read/write the main file and checksum file
+ private val threadPool = ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+
+ override def list(path: Path, filter: PathFilter): Array[FileStatus] = {
+ underlyingFileMgr.list(path, filter)
+ }
+
+ override def mkdirs(path: Path): Unit = {
+ underlyingFileMgr.mkdirs(path)
+ }
+
+ override def createAtomic(path: Path,
+ overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
+ assert(!isChecksumFile(path), "Cannot directly create a checksum file")
+
+ val mainFileFuture = Future {
+ underlyingFileMgr.createAtomic(path, overwriteIfPossible)
+ }(threadPool)
+
+ val checksumFileFuture = Future {
+ underlyingFileMgr.createAtomic(getChecksumPath(path), overwriteIfPossible)
+ }(threadPool)
+
+ new ChecksumCancellableFSDataOutputStream(
+ awaitResult(mainFileFuture, Duration.Inf),
+ path,
+ awaitResult(checksumFileFuture, Duration.Inf),
+ threadPool
+ )
+ }
+
+ override def open(path: Path): FSDataInputStream = {
+ assert(!isChecksumFile(path), "Cannot directly open a checksum file")
+
+ val checksumInputStreamFuture = Future {
+ try {
+ Some(underlyingFileMgr.open(getChecksumPath(path)))
+ } catch {
+ // In case the client previously had file checksum disabled.
+ // Then previously created files won't have checksum.
+ case _: FileNotFoundException =>
+ logWarning(log"No checksum file found for ${MDC(PATH, path)}, " +
+ log"hence no checksum verification.")
+ None
+ }
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.open(path)
+ }(threadPool)
+
+ val mainStream = awaitResult(mainInputStreamFuture, Duration.Inf)
+ val checksumStream = awaitResult(checksumInputStreamFuture, Duration.Inf)
+
+ checksumStream.map { chkStream =>
+ new ChecksumFSDataInputStream(mainStream, path, chkStream, threadPool)
+ }.getOrElse(mainStream)
+ }
+
+ override def exists(path: Path): Boolean = underlyingFileMgr.exists(path)
+
+ override def delete(path: Path): Unit = {
+ // Allowing directly deleting the checksum file for orphan checksum file scenario
+ if (isChecksumFile(path)) {
+ deleteChecksumFile(path)
+ } else if (allowConcurrentDelete) {
+ // Ideally, we should first try to delete the checksum file
+ // before the main file, to avoid a situation where the main file is deleted but the
+ // checksum file deletion failed. The client might not call delete again if the main file
+ // no longer exists.
+ // But if allowConcurrentDelete is enabled, then we can do it concurrently for perf.
+ // But the client would be responsible for cleaning up potential orphan checksum files
+ // if it happens.
+ val checksumInputStreamFuture = Future {
+ deleteChecksumFile(getChecksumPath(path))
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.delete(path)
+ }(threadPool)
+
+ awaitResult(mainInputStreamFuture, Duration.Inf)
+ awaitResult(checksumInputStreamFuture, Duration.Inf)
+ } else {
+ // First delete the checksum file, then main file
+ deleteChecksumFile(getChecksumPath(path))
+ underlyingFileMgr.delete(path)
+ }
+ }
+
+ private def deleteChecksumFile(checksumPath: Path): Unit = {
+ try {
+ underlyingFileMgr.delete(checksumPath)
+ logDebug(log"Deleted checksum file ${MDC(PATH, checksumPath)}")
+ } catch {
+ case _: FileNotFoundException =>
+ // Ignore if file has already been deleted
+ // or the main file was created initially without checksum
+ logWarning(log"Skipping deletion of checksum file ${MDC(PATH, checksumPath)} " +
+ log"since it does not exist.")
+ }
+ }
+
+ override def isLocal: Boolean = underlyingFileMgr.isLocal
+
+ override def createCheckpointDirectory(): Path = {
+ underlyingFileMgr.createCheckpointDirectory()
+ }
+
+ override def close(): Unit = {
+ threadPool.shutdown()
+ // Wait a bit for it to finish up in case there is any ongoing work
+ // Can consider making this timeout configurable, if needed
+ val timeoutMs = 100
+ if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
+ logWarning(log"Thread pool did not shutdown after ${MDC(TIMEOUT, timeoutMs)} ms")
+ }
+ }
+}
+
+private[streaming] object ChecksumCheckpointFileManager {
+ val CHECKSUM_FILE_SUFFIX = ".crc"
+
+ def awaitResult[T](future: Future[T], atMost: Duration): T = {
+ try {
+ ThreadUtils.awaitResult(future, atMost)
+ } catch {
+ // awaitResult wraps the exception. Unwrap it, and throw the actual error
+ case e: SparkException if e.getMessage.contains("Exception thrown in awaitResult") =>
+ throw e.getCause
+ }
+ }
+
+ def getChecksumPath(mainFilePath: Path): Path = {
+ new Path(mainFilePath.toString + CHECKSUM_FILE_SUFFIX)
+ }
+
+ def isChecksumFile(path: Path): Boolean = {
+ path.getName.endsWith(CHECKSUM_FILE_SUFFIX)
+ }
+}
+
+/** An implementation of [[FSDataInputStream]] that calculates the checksum of the file
+ * that the client is reading (main file) incrementally, while it is being read.
+ * It then does checksum verification on close, to verify that the computed checksum
+ * matches the expected checksum in the checksum file.
+ *
+ * Computing the checksum incrementally and doing the verification after file read is complete
+ * is for better performance, instead of first reading the entire file and doing verification
+ * before the client starts reading the file.
+ *
+ * @param mainStream Input stream for the main file the client wants to read
+ * @param path The path of the main file
+ * @param expectedChecksumStream The input stream for the checksum file
+ * @param threadPool Thread pool to use for concurrently operating on the main and checksum file
+ * */
+class ChecksumFSDataInputStream(
+ private val mainStream: FSDataInputStream,
+ path: Path,
+ private val expectedChecksumStream: FSDataInputStream,
+ private val threadPool: ExecutionContext)
+ extends FSDataInputStream(new CheckedSequentialInputStream(mainStream)) with Logging {
+
+ import ChecksumCheckpointFileManager._
+
+ @volatile private var verified = false
+ @volatile private var closed = false
+
+ override def close(): Unit = {
+ if (!closed) {
+ // We verify the checksum only when the client is done reading.
+ verifyChecksum()
+ closeInternal()
+ }
+ }
+
+ /** This is used to skip checksum verification on close.
+ * Avoid using this, and it is only used for a situation where the file is opened, read,
+ * then closed multiple times, and we want to avoid doing verification each time
+ * and only want to do it once.
+ * */
+ def closeWithoutChecksumVerification(): Unit = {
+ if (!closed) {
+ // Ideally this should be warning, but if a file is doing this frequently
+ // it will cause unnecessary noise in the logs. This can be changed later.
+ logDebug(log"Closing file ${MDC(PATH, path)} without checksum verification")
+ closeInternal()
+ }
+ }
+
+ private def closeInternal(): Unit = {
+ closed = true
+
+ val mainCloseFuture = Future {
+ super.close() // close the main file
+ }(threadPool)
+
+ val checksumCloseFuture = Future {
+ expectedChecksumStream.close() // close the checksum file
+ }(threadPool)
+
+ awaitResult(mainCloseFuture, Duration.Inf)
+ awaitResult(checksumCloseFuture, Duration.Inf)
+ }
+
+ private def verifyChecksum(): Unit = {
+ if (!verified) {
+ // It is possible the file was not read till the end by the reader,
+ // but we need the entire file content for checksum verification.
+ // Hence, we will read the file till the end from where reader stopped.
+ var remainingBytesRead = 0
+ val buffer = new Array[Byte](1024)
+ var bytesRead = 0
+ while ({ bytesRead = super.read(buffer); bytesRead != -1 }) {
+ remainingBytesRead += bytesRead
+ }
+
+ // we are at the end position so that tells us the size
+ val computedFileSize = mainStream.getPos
+
+ if (remainingBytesRead > 0) {
+ // Making this debug log since most of our files are not read exactly to the end
+ // and don't want to cause unnecessary noise in the logs.
+ logDebug(log"File ${MDC(PATH, path)} was not read till the end by reader. " +
+ log"Finished reading the rest of the file for checksum verification. " +
+ log"Remaining bytes read: ${MDC(NUM_BYTES, remainingBytesRead)}, " +
+ log"total size: ${MDC(NUM_BYTES, computedFileSize)}.")
+ }
+
+ // Read the expected checksum from the checksum file
+ val expectedChecksumJson = Source.fromInputStream(
+ expectedChecksumStream, StandardCharsets.UTF_8.name()).mkString
+ val expectedChecksum = Checksum.fromJson(expectedChecksumJson)
+ // Get what we computed while the main file was being read locally
+ // `in` is the CheckedSequentialInputStream we created
+ val computedChecksumValue = in.asInstanceOf[CheckedInputStream].getChecksum.getValue.toInt
+
+ logInfo(log"Verifying checksum for file ${MDC(PATH, path)}, " +
+ log"remainingBytesRead= ${MDC(NUM_BYTES, remainingBytesRead)}. " +
+ log"Computed(checksum= ${MDC(CHECKSUM, computedChecksumValue)}, " +
+ log"fileSize= ${MDC(NUM_BYTES, computedFileSize)})." +
+ log"Checksum file content: ${MDC(CHECKSUM, expectedChecksumJson)}")
+
+ verified = true
+
+ // Compare file size too, in case of collision
+ if (expectedChecksum.value.toInt != computedChecksumValue ||
+ expectedChecksum.mainFileSize != computedFileSize) {
+ throw QueryExecutionErrors.checkpointFileChecksumVerificationFailed(
Review Comment:
I don't think we should mix this layer with state store code. File manager is at a lower layer and this can be used for non-state store files too.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072017958
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1459,6 +1472,193 @@ abstract class StateStoreSuiteBase[ProviderClass <: StateStoreProvider]
}
}
+ testWithAllCodec("file checksum can be enabled and disabled for the same checkpoint") {
+ _ =>
Review Comment:
nit : move to line above ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072018624
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1459,6 +1472,193 @@ abstract class StateStoreSuiteBase[ProviderClass <: StateStoreProvider]
}
}
+ testWithAllCodec("file checksum can be enabled and disabled for the same checkpoint") {
+ _ =>
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ var version = 0L
+
+ // Commit to store using file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val store = provider.getStore(version)
+ put(store, "1", 11, 100)
+ put(store, "2", 22, 200)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit without file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 1)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+
+ put(store, "3", 33, 300)
+ put(store, "4", 44, 400)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit with file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 2)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+ assert(get(store, "3", 33) === Some(300))
+ assert(get(store, "4", 44) === Some(400))
+
+ put(store, "5", 55, 500)
+ version = store.commit()
+ }
+ }
+ }
+
+ test("checksum files are also cleaned up during maintenance") {
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ val numBatches = 6
+ val minDeltas = 2
+ // Adding 1 to ensure snapshot is uploaded.
+ // Snapshot upload might happen at minDeltas or minDeltas + 1, depending on the provider
+ val maintFrequency = minDeltas + 1
+ var version = 0L
+
+ withSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> minDeltas.toString,
+ SQLConf.MIN_BATCHES_TO_RETAIN.key -> "1",
+ // So that RocksDB will also generate changelog files
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" ->
+ true.toString) {
+
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (version + 1 to numBatches).foreach { i =>
+ version = putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = i % maintFrequency == 0)
+ }
+
+ // This is because, hdfs and rocksdb old files detection logic is different
+ provider match {
+ case _: HDFSBackedStateStoreProvider =>
+ // For HDFS State store, files left:
+ // 3.delta to 6.delta (+ checksum file)
+ // 3.snapshot (+ checksum file), 6.snapshot (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 12, expectedNumChecksumFiles = 6)
+ case _ =>
+ // For RocksDB State store, files left:
+ // 6.changelog (+ checksum file), 6.zip (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 4, expectedNumChecksumFiles = 2)
+ }
+ }
+
+ // turn off file checksum, and verify that the previously created checksum files
+ // will be deleted by maintenance
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (version + 1 to version + numBatches).foreach { i =>
+ version = putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = i % maintFrequency == 0)
+ }
+
+ // now verify no checksum files are left
+ // This is because, hdfs and rocksdb old files detection logic is different
+ provider match {
+ case _: HDFSBackedStateStoreProvider =>
+ // For HDFS State store, files left:
+ // 6.delta, 9.delta to 12.delta
+ // 9.snapshot, 12.snapshot
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 7, expectedNumChecksumFiles = 0)
+ case _ =>
+ // For RocksDB State store, files left:
+ // 12.changelog, 12.zip
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 2, expectedNumChecksumFiles = 0)
+ }
+ }
+ }
+ }
+ }
+
+ testWithAllCodec("overwrite state file without overwriting checksum file") { _ =>
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ val numBatches = 3
+ val minDeltas = 2
+
+ withSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> minDeltas.toString,
+ // So that RocksDB will also generate changelog files
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" ->
+ true.toString) {
+
+ // First run with file checksum enabled. It will generate state and checksum files.
+ // Turn off file checksum, and regenerate only the state files
+ Seq(true, false).foreach { fileChecksumEnabled =>
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> fileChecksumEnabled.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (1 to numBatches).foreach { i =>
+ putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = false)
+ }
+
+ // This should only create snapshot and no delete
+ provider.doMaintenance()
+
+ // number of files should be the same.
+ // 3 delta/changelog files, 1 snapshot (with checksum files)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 8, expectedNumChecksumFiles = 4)
+ }
+ }
+ }
+
+ withSQLConf(
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ // now try to load the store with checksum enabled.
+ // It will verify the overwritten state files with the checksum files.
+ (1 to numBatches).foreach { i =>
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ // load from DFS should be successful
+ val store = provider.getStore(i)
+ store.abort()
+ }
+ }
+ }
+ }
+ }
+
+ private def verifyChecksumFiles(
Review Comment:
Are we covering all the files we are adding the checksum for ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072006492
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,512 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used
+ value: String,
Review Comment:
now using the original int, we can use byte array later when we have other algos, in order to be agnostic. Added comment for that
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2074772335
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,513 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used and be easily readable.
+ // We can change this to byte array later if we start using algos with large values.
+ value: String,
+ mainFileSize: Long,
+ timestampMs: Long,
+ creator: ChecksumFileCreatorInfo) {
+
+ import Checksum._
+
+ def json(): String = {
+ mapper.writeValueAsString(this)
+ }
+}
+
+object Checksum {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ /** Used to convert between class and JSON. */
+ lazy val mapper = {
+ val _mapper = new ObjectMapper with ClassTagExtensions
+ _mapper.setSerializationInclusion(Include.NON_ABSENT)
+ _mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ _mapper.registerModule(DefaultScalaModule)
+ _mapper
+ }
+
+ def fromJson(json: String): Checksum = {
+ Serialization.read[Checksum](json)
+ }
+}
+
+/** Holds the path of the checksum file and knows the main file path. */
+case class ChecksumFile(path: Path) {
+ import ChecksumCheckpointFileManager._
+ assert(isChecksumFile(path), "path is not a checksum file")
+
+ val mainFilePath = new Path(path.toString.stripSuffix(CHECKSUM_FILE_SUFFIX))
+
+ /** The name of the file without any extensions e.g. my-file.txt.crc returns my-file */
+ val baseName: String = path.getName.split("\\.").head
+}
+
+/**
+ * A [[CheckpointFileManager]] that creates a checksum file for the main file.
+ * This wraps another [[CheckpointFileManager]] and adds checksum functionality on top of it.
+ * Under the hood, when a file is created, it also creates a checksum file with the same name as
+ * the main file but adds a suffix. It returns [[ChecksumCancellableFSDataOutputStream]]
+ * which handles the writing of the main file and checksum file.
+ *
+ * When a file is opened, it returns [[ChecksumFSDataInputStream]], which handles reading
+ * the main file and checksum file and does the checksum verification.
+ *
+ * In order to reduce the impact of reading/writing 2 files instead of 1, it uses a threadpool
+ * to read/write both files concurrently.
+ *
+ * @note
+ * It is able to read files written by other [[CheckpointFileManager]], that don't have checksum.
+ * It automatically deletes the checksum file when the main file is deleted.
+ * If you delete the main file with a different type of manager, then the checksum file will be
+ * left behind (i.e. orphan checksum file), since they don't know about it. It would be your
+ * responsibility to delete the orphan checksum files.
+ *
+ * @param underlyingFileMgr The file manager to use under the hood
+ * @param allowConcurrentDelete If true, allows deleting the main and checksum file concurrently.
+ * This is a perf optimization, but can potentially lead to
+ * orphan checksum files. If using this, it is your responsibility
+ * to clean up the potential orphan checksum files.
+ * @param numThreads This is the number of threads to use for the thread pool, for reading/writing
+ * files. To avoid blocking, if the file manager instance is being used by a
+ * single thread, then you can set this to 2 (one thread for main file, another
+ * for checksum file).
+ * If file manager is shared by multiple threads, you can set it to
+ * number of threads using file manager * 2.
+ * Setting this differently can lead to file operation being blocked waiting for
+ * a free thread.
+ */
+class ChecksumCheckpointFileManager(
+ private val underlyingFileMgr: CheckpointFileManager,
+ val allowConcurrentDelete: Boolean = false,
+ val numThreads: Int)
+ extends CheckpointFileManager with Logging {
+ assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
+ "and another for the checksum file")
+
+ import ChecksumCheckpointFileManager._
+
+ // This allows us to concurrently read/write the main file and checksum file
+ private val threadPool = ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+
+ override def list(path: Path, filter: PathFilter): Array[FileStatus] = {
+ underlyingFileMgr.list(path, filter)
+ }
+
+ override def mkdirs(path: Path): Unit = {
+ underlyingFileMgr.mkdirs(path)
+ }
+
+ override def createAtomic(path: Path,
+ overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
+ assert(!isChecksumFile(path), "Cannot directly create a checksum file")
+
+ val mainFileFuture = Future {
+ underlyingFileMgr.createAtomic(path, overwriteIfPossible)
+ }(threadPool)
+
+ val checksumFileFuture = Future {
+ underlyingFileMgr.createAtomic(getChecksumPath(path), overwriteIfPossible)
+ }(threadPool)
+
+ new ChecksumCancellableFSDataOutputStream(
+ awaitResult(mainFileFuture, Duration.Inf),
+ path,
+ awaitResult(checksumFileFuture, Duration.Inf),
+ threadPool
+ )
+ }
+
+ override def open(path: Path): FSDataInputStream = {
+ assert(!isChecksumFile(path), "Cannot directly open a checksum file")
+
+ val checksumInputStreamFuture = Future {
+ try {
+ Some(underlyingFileMgr.open(getChecksumPath(path)))
+ } catch {
+ // In case the client previously had file checksum disabled.
+ // Then previously created files won't have checksum.
+ case _: FileNotFoundException =>
+ logWarning(log"No checksum file found for ${MDC(PATH, path)}, " +
+ log"hence no checksum verification.")
+ None
+ }
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.open(path)
+ }(threadPool)
+
+ val mainStream = awaitResult(mainInputStreamFuture, Duration.Inf)
+ val checksumStream = awaitResult(checksumInputStreamFuture, Duration.Inf)
+
+ checksumStream.map { chkStream =>
+ new ChecksumFSDataInputStream(mainStream, path, chkStream, threadPool)
+ }.getOrElse(mainStream)
+ }
+
+ override def exists(path: Path): Boolean = underlyingFileMgr.exists(path)
+
+ override def delete(path: Path): Unit = {
+ // Allowing directly deleting the checksum file for orphan checksum file scenario
+ if (isChecksumFile(path)) {
+ deleteChecksumFile(path)
+ } else if (allowConcurrentDelete) {
+ // Ideally, we should first try to delete the checksum file
+ // before the main file, to avoid a situation where the main file is deleted but the
+ // checksum file deletion failed. The client might not call delete again if the main file
+ // no longer exists.
+ // But if allowConcurrentDelete is enabled, then we can do it concurrently for perf.
+ // But the client would be responsible for cleaning up potential orphan checksum files
+ // if it happens.
+ val checksumInputStreamFuture = Future {
+ deleteChecksumFile(getChecksumPath(path))
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.delete(path)
+ }(threadPool)
+
+ awaitResult(mainInputStreamFuture, Duration.Inf)
+ awaitResult(checksumInputStreamFuture, Duration.Inf)
+ } else {
+ // First delete the checksum file, then main file
+ deleteChecksumFile(getChecksumPath(path))
+ underlyingFileMgr.delete(path)
+ }
+ }
+
+ private def deleteChecksumFile(checksumPath: Path): Unit = {
+ try {
+ underlyingFileMgr.delete(checksumPath)
+ logDebug(log"Deleted checksum file ${MDC(PATH, checksumPath)}")
+ } catch {
+ case _: FileNotFoundException =>
+ // Ignore if file has already been deleted
+ // or the main file was created initially without checksum
+ logWarning(log"Skipping deletion of checksum file ${MDC(PATH, checksumPath)} " +
+ log"since it does not exist.")
+ }
+ }
+
+ override def isLocal: Boolean = underlyingFileMgr.isLocal
+
+ override def createCheckpointDirectory(): Path = {
+ underlyingFileMgr.createCheckpointDirectory()
+ }
+
+ override def close(): Unit = {
+ threadPool.shutdown()
+ // Wait a bit for it to finish up in case there is any ongoing work
+ // Can consider making this timeout configurable, if needed
+ val timeoutMs = 100
+ if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
+ logWarning(log"Thread pool did not shutdown after ${MDC(TIMEOUT, timeoutMs)} ms")
+ }
+ }
+}
+
+private[streaming] object ChecksumCheckpointFileManager {
+ val CHECKSUM_FILE_SUFFIX = ".crc"
+
+ def awaitResult[T](future: Future[T], atMost: Duration): T = {
+ try {
+ ThreadUtils.awaitResult(future, atMost)
+ } catch {
+ // awaitResult wraps the exception. Unwrap it, and throw the actual error
+ case e: SparkException if e.getMessage.contains("Exception thrown in awaitResult") =>
+ throw e.getCause
+ }
+ }
+
+ def getChecksumPath(mainFilePath: Path): Path = {
+ new Path(mainFilePath.toString + CHECKSUM_FILE_SUFFIX)
+ }
+
+ def isChecksumFile(path: Path): Boolean = {
+ path.getName.endsWith(CHECKSUM_FILE_SUFFIX)
+ }
+}
+
+/** An implementation of [[FSDataInputStream]] that calculates the checksum of the file
+ * that the client is reading (main file) incrementally, while it is being read.
+ * It then does checksum verification on close, to verify that the computed checksum
+ * matches the expected checksum in the checksum file.
+ *
+ * Computing the checksum incrementally and doing the verification after file read is complete
+ * is for better performance, instead of first reading the entire file and doing verification
+ * before the client starts reading the file.
+ *
+ * @param mainStream Input stream for the main file the client wants to read
+ * @param path The path of the main file
+ * @param expectedChecksumStream The input stream for the checksum file
+ * @param threadPool Thread pool to use for concurrently operating on the main and checksum file
+ * */
+class ChecksumFSDataInputStream(
+ private val mainStream: FSDataInputStream,
+ path: Path,
+ private val expectedChecksumStream: FSDataInputStream,
+ private val threadPool: ExecutionContext)
+ extends FSDataInputStream(new CheckedSequentialInputStream(mainStream)) with Logging {
+
+ import ChecksumCheckpointFileManager._
+
+ @volatile private var verified = false
+ @volatile private var closed = false
+
+ override def close(): Unit = {
+ if (!closed) {
+ // We verify the checksum only when the client is done reading.
+ verifyChecksum()
+ closeInternal()
+ }
+ }
+
+ /** This is used to skip checksum verification on close.
+ * Avoid using this, and it is only used for a situation where the file is opened, read,
+ * then closed multiple times, and we want to avoid doing verification each time
+ * and only want to do it once.
+ * */
+ def closeWithoutChecksumVerification(): Unit = {
+ if (!closed) {
+ // Ideally this should be warning, but if a file is doing this frequently
+ // it will cause unnecessary noise in the logs. This can be changed later.
+ logDebug(log"Closing file ${MDC(PATH, path)} without checksum verification")
+ closeInternal()
+ }
+ }
+
+ private def closeInternal(): Unit = {
+ closed = true
+
+ val mainCloseFuture = Future {
+ super.close() // close the main file
+ }(threadPool)
+
+ val checksumCloseFuture = Future {
+ expectedChecksumStream.close() // close the checksum file
+ }(threadPool)
+
+ awaitResult(mainCloseFuture, Duration.Inf)
+ awaitResult(checksumCloseFuture, Duration.Inf)
+ }
+
+ private def verifyChecksum(): Unit = {
+ if (!verified) {
+ // It is possible the file was not read till the end by the reader,
+ // but we need the entire file content for checksum verification.
+ // Hence, we will read the file till the end from where reader stopped.
+ var remainingBytesRead = 0
+ val buffer = new Array[Byte](1024)
+ var bytesRead = 0
+ while ({ bytesRead = super.read(buffer); bytesRead != -1 }) {
+ remainingBytesRead += bytesRead
+ }
+
+ // we are at the end position so that tells us the size
+ val computedFileSize = mainStream.getPos
+
+ if (remainingBytesRead > 0) {
+ // Making this debug log since most of our files are not read exactly to the end
+ // and don't want to cause unnecessary noise in the logs.
+ logDebug(log"File ${MDC(PATH, path)} was not read till the end by reader. " +
+ log"Finished reading the rest of the file for checksum verification. " +
+ log"Remaining bytes read: ${MDC(NUM_BYTES, remainingBytesRead)}, " +
+ log"total size: ${MDC(NUM_BYTES, computedFileSize)}.")
+ }
+
+ // Read the expected checksum from the checksum file
+ val expectedChecksumJson = Source.fromInputStream(
+ expectedChecksumStream, StandardCharsets.UTF_8.name()).mkString
+ val expectedChecksum = Checksum.fromJson(expectedChecksumJson)
+ // Get what we computed while the main file was being read locally
+ // `in` is the CheckedSequentialInputStream we created
+ val computedChecksumValue = in.asInstanceOf[CheckedInputStream].getChecksum.getValue.toInt
+
+ logInfo(log"Verifying checksum for file ${MDC(PATH, path)}, " +
+ log"remainingBytesRead= ${MDC(NUM_BYTES, remainingBytesRead)}. " +
+ log"Computed(checksum= ${MDC(CHECKSUM, computedChecksumValue)}, " +
+ log"fileSize= ${MDC(NUM_BYTES, computedFileSize)})." +
+ log"Checksum file content: ${MDC(CHECKSUM, expectedChecksumJson)}")
+
+ verified = true
+
+ // Compare file size too, in case of collision
+ if (expectedChecksum.value.toInt != computedChecksumValue ||
+ expectedChecksum.mainFileSize != computedFileSize) {
+ throw QueryExecutionErrors.checkpointFileChecksumVerificationFailed(
+ path,
+ expectedSize = expectedChecksum.mainFileSize,
+ expectedChecksum.value,
+ computedSize = computedFileSize,
+ computedChecksumValue.toString)
+ }
+ }
+ }
+}
+
+/** This implements [[CheckedInputStream]] that allows us to compute the checksum as the file
+ * is being read. [[FSDataInputStream]] needs the passed in input stream to implement Seekable
+ * and PositionedReadable. We want the file to be read sequentially only to allow us to correctly
+ * compute the checksum. This is blocking the seekable apis from being used in the underlying stream
+ * */
+private class CheckedSequentialInputStream(data: InputStream)
+ // TODO: Make the checksum algo configurable
+ extends CheckedInputStream(data, new CRC32C())
+ with Seekable
+ with PositionedReadable {
+
+ // Seekable methods
+ override def seek(pos: Long): Unit = {
+ throw new UnsupportedOperationException("Seek not supported")
+ }
+
+ override def getPos: Long = {
+ throw new UnsupportedOperationException("getPos not supported")
+ }
+
+ override def seekToNewSource(targetPos: Long): Boolean = {
+ throw new UnsupportedOperationException("seekToNewSource not supported")
+ }
+
+ // PositionedReadable methods
+ def read(position: Long, buffer: Array[Byte], offset: Int, length: Int): Int = {
+ throw new UnsupportedOperationException("read from position not supported")
+ }
+
+ def readFully(position: Long, buffer: Array[Byte], offset: Int, length: Int): Unit = {
+ throw new UnsupportedOperationException("readFully from position not supported")
+ }
+
+ def readFully(position: Long, buffer: Array[Byte]): Unit = {
+ throw new UnsupportedOperationException("readFully from position not supported")
+ }
+}
+
+/** An implementation of [[CancellableFSDataOutputStream]] that calculates the checksum of the file
+ * that the client is writing (main file) incrementally, while it is being written.
+ * It then writes the main file and an additional checksum file, which will be used for verification
+ * by [[ChecksumFSDataInputStream]] on file read.
+ *
+ * @param mainStream Output stream for the main file the client wants to write to
+ * @param path The path of the main file
+ * @param checksumStream Output stream for the checksum file to write the computed checksum
+ * @param uploadThreadPool Thread pool used to concurrently upload the main and checksum file
+ * */
+class ChecksumCancellableFSDataOutputStream(
+ private val mainStream: CancellableFSDataOutputStream,
+ path: Path,
+ private val checksumStream: CancellableFSDataOutputStream,
+ private val uploadThreadPool: ExecutionContext)
+ // TODO: make the checksum algo configurable
Review Comment:
done
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2074772803
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1459,6 +1472,193 @@ abstract class StateStoreSuiteBase[ProviderClass <: StateStoreProvider]
}
}
+ testWithAllCodec("file checksum can be enabled and disabled for the same checkpoint") {
+ _ =>
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ var version = 0L
+
+ // Commit to store using file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val store = provider.getStore(version)
+ put(store, "1", 11, 100)
+ put(store, "2", 22, 200)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit without file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 1)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+
+ put(store, "3", 33, 300)
+ put(store, "4", 44, 400)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit with file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 2)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+ assert(get(store, "3", 33) === Some(300))
+ assert(get(store, "4", 44) === Some(400))
+
+ put(store, "5", 55, 500)
+ version = store.commit()
+ }
+ }
+ }
+
+ test("checksum files are also cleaned up during maintenance") {
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ val numBatches = 6
+ val minDeltas = 2
+ // Adding 1 to ensure snapshot is uploaded.
+ // Snapshot upload might happen at minDeltas or minDeltas + 1, depending on the provider
+ val maintFrequency = minDeltas + 1
+ var version = 0L
+
+ withSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> minDeltas.toString,
+ SQLConf.MIN_BATCHES_TO_RETAIN.key -> "1",
+ // So that RocksDB will also generate changelog files
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" ->
+ true.toString) {
+
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (version + 1 to numBatches).foreach { i =>
+ version = putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = i % maintFrequency == 0)
+ }
+
+ // This is because, hdfs and rocksdb old files detection logic is different
+ provider match {
+ case _: HDFSBackedStateStoreProvider =>
+ // For HDFS State store, files left:
+ // 3.delta to 6.delta (+ checksum file)
+ // 3.snapshot (+ checksum file), 6.snapshot (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 12, expectedNumChecksumFiles = 6)
+ case _ =>
+ // For RocksDB State store, files left:
+ // 6.changelog (+ checksum file), 6.zip (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 4, expectedNumChecksumFiles = 2)
+ }
+ }
+
+ // turn off file checksum, and verify that the previously created checksum files
+ // will be deleted by maintenance
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
Review Comment:
added golden files test
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072017394
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1459,6 +1472,193 @@ abstract class StateStoreSuiteBase[ProviderClass <: StateStoreProvider]
}
}
+ testWithAllCodec("file checksum can be enabled and disabled for the same checkpoint") {
+ _ =>
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ var version = 0L
+
+ // Commit to store using file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val store = provider.getStore(version)
+ put(store, "1", 11, 100)
+ put(store, "2", 22, 200)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit without file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 1)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+
+ put(store, "3", 33, 300)
+ put(store, "4", 44, 400)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit with file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 2)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+ assert(get(store, "3", 33) === Some(300))
+ assert(get(store, "4", 44) === Some(400))
+
+ put(store, "5", 55, 500)
+ version = store.commit()
+ }
+ }
+ }
+
+ test("checksum files are also cleaned up during maintenance") {
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ val numBatches = 6
+ val minDeltas = 2
+ // Adding 1 to ensure snapshot is uploaded.
+ // Snapshot upload might happen at minDeltas or minDeltas + 1, depending on the provider
+ val maintFrequency = minDeltas + 1
+ var version = 0L
+
+ withSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> minDeltas.toString,
+ SQLConf.MIN_BATCHES_TO_RETAIN.key -> "1",
+ // So that RocksDB will also generate changelog files
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" ->
+ true.toString) {
+
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (version + 1 to numBatches).foreach { i =>
+ version = putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = i % maintFrequency == 0)
+ }
+
+ // This is because, hdfs and rocksdb old files detection logic is different
+ provider match {
+ case _: HDFSBackedStateStoreProvider =>
+ // For HDFS State store, files left:
+ // 3.delta to 6.delta (+ checksum file)
+ // 3.snapshot (+ checksum file), 6.snapshot (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 12, expectedNumChecksumFiles = 6)
+ case _ =>
+ // For RocksDB State store, files left:
+ // 6.changelog (+ checksum file), 6.zip (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 4, expectedNumChecksumFiles = 2)
+ }
+ }
+
+ // turn off file checksum, and verify that the previously created checksum files
+ // will be deleted by maintenance
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
Review Comment:
Can we also test on golden files created from a Spark 4.0 checkpoint ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072053570
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreSuite.scala:
##########
@@ -1459,6 +1472,193 @@ abstract class StateStoreSuiteBase[ProviderClass <: StateStoreProvider]
}
}
+ testWithAllCodec("file checksum can be enabled and disabled for the same checkpoint") {
+ _ =>
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ var version = 0L
+
+ // Commit to store using file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ val store = provider.getStore(version)
+ put(store, "1", 11, 100)
+ put(store, "2", 22, 200)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit without file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 1)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+
+ put(store, "3", 33, 300)
+ put(store, "4", 44, 400)
+ version = store.commit()
+ }
+ }
+
+ // Reload the store and commit with file checksum
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> true.toString) {
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ assert(version == 2)
+ val store = provider.getStore(version)
+ assert(get(store, "1", 11) === Some(100))
+ assert(get(store, "2", 22) === Some(200))
+ assert(get(store, "3", 33) === Some(300))
+ assert(get(store, "4", 44) === Some(400))
+
+ put(store, "5", 55, 500)
+ version = store.commit()
+ }
+ }
+ }
+
+ test("checksum files are also cleaned up during maintenance") {
+ val storeId = StateStoreId(newDir(), 0L, 1)
+ val numBatches = 6
+ val minDeltas = 2
+ // Adding 1 to ensure snapshot is uploaded.
+ // Snapshot upload might happen at minDeltas or minDeltas + 1, depending on the provider
+ val maintFrequency = minDeltas + 1
+ var version = 0L
+
+ withSQLConf(SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> minDeltas.toString,
+ SQLConf.MIN_BATCHES_TO_RETAIN.key -> "1",
+ // So that RocksDB will also generate changelog files
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX + ".changelogCheckpointing.enabled" ->
+ true.toString) {
+
+ tryWithProviderResource(newStoreProviderWithClonedConf(storeId)) { provider =>
+ (version + 1 to numBatches).foreach { i =>
+ version = putAndCommitStore(
+ provider, loadVersion = i - 1, doMaintenance = i % maintFrequency == 0)
+ }
+
+ // This is because, hdfs and rocksdb old files detection logic is different
+ provider match {
+ case _: HDFSBackedStateStoreProvider =>
+ // For HDFS State store, files left:
+ // 3.delta to 6.delta (+ checksum file)
+ // 3.snapshot (+ checksum file), 6.snapshot (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 12, expectedNumChecksumFiles = 6)
+ case _ =>
+ // For RocksDB State store, files left:
+ // 6.changelog (+ checksum file), 6.zip (+ checksum file)
+ verifyChecksumFiles(storeId.storeCheckpointLocation().toString,
+ expectedNumFiles = 4, expectedNumChecksumFiles = 2)
+ }
+ }
+
+ // turn off file checksum, and verify that the previously created checksum files
+ // will be deleted by maintenance
+ withSQLConf(SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key -> false.toString) {
Review Comment:
Yea - but to be safe - we usually also confirm against released versions. so you can generate the golden files for that version and run your tests against that path (you can check the usage of this dir for eg - https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala#L1188 )
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072002173
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,513 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used and be easily readable.
+ // We can change this to byte array later if we start using algos with large values.
+ value: String,
+ mainFileSize: Long,
+ timestampMs: Long,
+ creator: ChecksumFileCreatorInfo) {
+
+ import Checksum._
+
+ def json(): String = {
+ mapper.writeValueAsString(this)
+ }
+}
+
+object Checksum {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ /** Used to convert between class and JSON. */
+ lazy val mapper = {
+ val _mapper = new ObjectMapper with ClassTagExtensions
+ _mapper.setSerializationInclusion(Include.NON_ABSENT)
+ _mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ _mapper.registerModule(DefaultScalaModule)
+ _mapper
+ }
+
+ def fromJson(json: String): Checksum = {
+ Serialization.read[Checksum](json)
+ }
+}
+
+/** Holds the path of the checksum file and knows the main file path. */
+case class ChecksumFile(path: Path) {
+ import ChecksumCheckpointFileManager._
+ assert(isChecksumFile(path), "path is not a checksum file")
+
+ val mainFilePath = new Path(path.toString.stripSuffix(CHECKSUM_FILE_SUFFIX))
+
+ /** The name of the file without any extensions e.g. my-file.txt.crc returns my-file */
+ val baseName: String = path.getName.split("\\.").head
+}
+
+/**
+ * A [[CheckpointFileManager]] that creates a checksum file for the main file.
+ * This wraps another [[CheckpointFileManager]] and adds checksum functionality on top of it.
+ * Under the hood, when a file is created, it also creates a checksum file with the same name as
+ * the main file but adds a suffix. It returns [[ChecksumCancellableFSDataOutputStream]]
+ * which handles the writing of the main file and checksum file.
+ *
+ * When a file is opened, it returns [[ChecksumFSDataInputStream]], which handles reading
+ * the main file and checksum file and does the checksum verification.
+ *
+ * In order to reduce the impact of reading/writing 2 files instead of 1, it uses a threadpool
+ * to read/write both files concurrently.
+ *
+ * @note
+ * It is able to read files written by other [[CheckpointFileManager]], that don't have checksum.
+ * It automatically deletes the checksum file when the main file is deleted.
+ * If you delete the main file with a different type of manager, then the checksum file will be
+ * left behind (i.e. orphan checksum file), since they don't know about it. It would be your
+ * responsibility to delete the orphan checksum files.
+ *
+ * @param underlyingFileMgr The file manager to use under the hood
+ * @param allowConcurrentDelete If true, allows deleting the main and checksum file concurrently.
+ * This is a perf optimization, but can potentially lead to
+ * orphan checksum files. If using this, it is your responsibility
+ * to clean up the potential orphan checksum files.
+ * @param numThreads This is the number of threads to use for the thread pool, for reading/writing
+ * files. To avoid blocking, if the file manager instance is being used by a
+ * single thread, then you can set this to 2 (one thread for main file, another
+ * for checksum file).
+ * If file manager is shared by multiple threads, you can set it to
+ * number of threads using file manager * 2.
+ * Setting this differently can lead to file operation being blocked waiting for
+ * a free thread.
+ */
+class ChecksumCheckpointFileManager(
+ private val underlyingFileMgr: CheckpointFileManager,
+ val allowConcurrentDelete: Boolean = false,
+ val numThreads: Int)
+ extends CheckpointFileManager with Logging {
+ assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
+ "and another for the checksum file")
+
+ import ChecksumCheckpointFileManager._
+
+ // This allows us to concurrently read/write the main file and checksum file
+ private val threadPool = ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+
+ override def list(path: Path, filter: PathFilter): Array[FileStatus] = {
+ underlyingFileMgr.list(path, filter)
+ }
+
+ override def mkdirs(path: Path): Unit = {
+ underlyingFileMgr.mkdirs(path)
+ }
+
+ override def createAtomic(path: Path,
+ overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
+ assert(!isChecksumFile(path), "Cannot directly create a checksum file")
+
+ val mainFileFuture = Future {
+ underlyingFileMgr.createAtomic(path, overwriteIfPossible)
+ }(threadPool)
+
+ val checksumFileFuture = Future {
+ underlyingFileMgr.createAtomic(getChecksumPath(path), overwriteIfPossible)
+ }(threadPool)
+
+ new ChecksumCancellableFSDataOutputStream(
+ awaitResult(mainFileFuture, Duration.Inf),
+ path,
+ awaitResult(checksumFileFuture, Duration.Inf),
+ threadPool
+ )
+ }
+
+ override def open(path: Path): FSDataInputStream = {
+ assert(!isChecksumFile(path), "Cannot directly open a checksum file")
+
+ val checksumInputStreamFuture = Future {
+ try {
+ Some(underlyingFileMgr.open(getChecksumPath(path)))
+ } catch {
+ // In case the client previously had file checksum disabled.
+ // Then previously created files won't have checksum.
+ case _: FileNotFoundException =>
+ logWarning(log"No checksum file found for ${MDC(PATH, path)}, " +
+ log"hence no checksum verification.")
+ None
+ }
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.open(path)
+ }(threadPool)
+
+ val mainStream = awaitResult(mainInputStreamFuture, Duration.Inf)
+ val checksumStream = awaitResult(checksumInputStreamFuture, Duration.Inf)
+
+ checksumStream.map { chkStream =>
+ new ChecksumFSDataInputStream(mainStream, path, chkStream, threadPool)
+ }.getOrElse(mainStream)
+ }
+
+ override def exists(path: Path): Boolean = underlyingFileMgr.exists(path)
+
+ override def delete(path: Path): Unit = {
+ // Allowing directly deleting the checksum file for orphan checksum file scenario
+ if (isChecksumFile(path)) {
+ deleteChecksumFile(path)
+ } else if (allowConcurrentDelete) {
+ // Ideally, we should first try to delete the checksum file
+ // before the main file, to avoid a situation where the main file is deleted but the
+ // checksum file deletion failed. The client might not call delete again if the main file
+ // no longer exists.
+ // But if allowConcurrentDelete is enabled, then we can do it concurrently for perf.
+ // But the client would be responsible for cleaning up potential orphan checksum files
+ // if it happens.
+ val checksumInputStreamFuture = Future {
+ deleteChecksumFile(getChecksumPath(path))
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.delete(path)
+ }(threadPool)
+
+ awaitResult(mainInputStreamFuture, Duration.Inf)
+ awaitResult(checksumInputStreamFuture, Duration.Inf)
+ } else {
+ // First delete the checksum file, then main file
+ deleteChecksumFile(getChecksumPath(path))
+ underlyingFileMgr.delete(path)
+ }
+ }
+
+ private def deleteChecksumFile(checksumPath: Path): Unit = {
+ try {
+ underlyingFileMgr.delete(checksumPath)
+ logDebug(log"Deleted checksum file ${MDC(PATH, checksumPath)}")
+ } catch {
+ case _: FileNotFoundException =>
+ // Ignore if file has already been deleted
+ // or the main file was created initially without checksum
+ logWarning(log"Skipping deletion of checksum file ${MDC(PATH, checksumPath)} " +
+ log"since it does not exist.")
+ }
+ }
+
+ override def isLocal: Boolean = underlyingFileMgr.isLocal
+
+ override def createCheckpointDirectory(): Path = {
+ underlyingFileMgr.createCheckpointDirectory()
+ }
+
+ override def close(): Unit = {
+ threadPool.shutdown()
+ // Wait a bit for it to finish up in case there is any ongoing work
+ // Can consider making this timeout configurable, if needed
+ val timeoutMs = 100
+ if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
Review Comment:
Should we wait a bit longer here ? also - should we force the shutdown on some timeout ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2076054719
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManagerSuite.scala:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io._
+import java.nio.charset.StandardCharsets
+
+import scala.io.Source
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs._
+
+import org.apache.spark.SparkException
+
+abstract class ChecksumCheckpointFileManagerSuite extends CheckpointFileManagerTestsOnLocalFs {
Review Comment:
Can we add a couple of lines to explain what tests are covered in this file ?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala:
##########
@@ -342,6 +363,8 @@ class RocksDBFileManager(
} else {
// Delete all non-immutable files in local dir, and unzip new ones from DFS commit file
listRocksDBFiles(localDir)._2.foreach(_.delete())
+ // TODO(SPARK-51988): We are using fs here to read the file, checksum verification
+ // wouldn't happen for the file if enabled, since it is not using fm to open it.
Review Comment:
nit: extra space at beginning ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2074772105
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,513 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used and be easily readable.
+ // We can change this to byte array later if we start using algos with large values.
+ value: String,
+ mainFileSize: Long,
+ timestampMs: Long,
+ creator: ChecksumFileCreatorInfo) {
+
+ import Checksum._
+
+ def json(): String = {
+ mapper.writeValueAsString(this)
+ }
+}
+
+object Checksum {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ /** Used to convert between class and JSON. */
+ lazy val mapper = {
+ val _mapper = new ObjectMapper with ClassTagExtensions
+ _mapper.setSerializationInclusion(Include.NON_ABSENT)
+ _mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ _mapper.registerModule(DefaultScalaModule)
+ _mapper
+ }
+
+ def fromJson(json: String): Checksum = {
+ Serialization.read[Checksum](json)
+ }
+}
+
+/** Holds the path of the checksum file and knows the main file path. */
+case class ChecksumFile(path: Path) {
+ import ChecksumCheckpointFileManager._
+ assert(isChecksumFile(path), "path is not a checksum file")
+
+ val mainFilePath = new Path(path.toString.stripSuffix(CHECKSUM_FILE_SUFFIX))
+
+ /** The name of the file without any extensions e.g. my-file.txt.crc returns my-file */
+ val baseName: String = path.getName.split("\\.").head
+}
+
+/**
+ * A [[CheckpointFileManager]] that creates a checksum file for the main file.
+ * This wraps another [[CheckpointFileManager]] and adds checksum functionality on top of it.
+ * Under the hood, when a file is created, it also creates a checksum file with the same name as
+ * the main file but adds a suffix. It returns [[ChecksumCancellableFSDataOutputStream]]
+ * which handles the writing of the main file and checksum file.
+ *
+ * When a file is opened, it returns [[ChecksumFSDataInputStream]], which handles reading
+ * the main file and checksum file and does the checksum verification.
+ *
+ * In order to reduce the impact of reading/writing 2 files instead of 1, it uses a threadpool
+ * to read/write both files concurrently.
+ *
+ * @note
+ * It is able to read files written by other [[CheckpointFileManager]], that don't have checksum.
+ * It automatically deletes the checksum file when the main file is deleted.
+ * If you delete the main file with a different type of manager, then the checksum file will be
+ * left behind (i.e. orphan checksum file), since they don't know about it. It would be your
+ * responsibility to delete the orphan checksum files.
+ *
+ * @param underlyingFileMgr The file manager to use under the hood
+ * @param allowConcurrentDelete If true, allows deleting the main and checksum file concurrently.
+ * This is a perf optimization, but can potentially lead to
+ * orphan checksum files. If using this, it is your responsibility
+ * to clean up the potential orphan checksum files.
+ * @param numThreads This is the number of threads to use for the thread pool, for reading/writing
+ * files. To avoid blocking, if the file manager instance is being used by a
+ * single thread, then you can set this to 2 (one thread for main file, another
+ * for checksum file).
+ * If file manager is shared by multiple threads, you can set it to
+ * number of threads using file manager * 2.
+ * Setting this differently can lead to file operation being blocked waiting for
+ * a free thread.
+ */
+class ChecksumCheckpointFileManager(
+ private val underlyingFileMgr: CheckpointFileManager,
+ val allowConcurrentDelete: Boolean = false,
+ val numThreads: Int)
+ extends CheckpointFileManager with Logging {
+ assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
+ "and another for the checksum file")
+
+ import ChecksumCheckpointFileManager._
+
+ // This allows us to concurrently read/write the main file and checksum file
+ private val threadPool = ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+
+ override def list(path: Path, filter: PathFilter): Array[FileStatus] = {
+ underlyingFileMgr.list(path, filter)
+ }
+
+ override def mkdirs(path: Path): Unit = {
+ underlyingFileMgr.mkdirs(path)
+ }
+
+ override def createAtomic(path: Path,
+ overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
+ assert(!isChecksumFile(path), "Cannot directly create a checksum file")
+
+ val mainFileFuture = Future {
+ underlyingFileMgr.createAtomic(path, overwriteIfPossible)
+ }(threadPool)
+
+ val checksumFileFuture = Future {
+ underlyingFileMgr.createAtomic(getChecksumPath(path), overwriteIfPossible)
+ }(threadPool)
+
+ new ChecksumCancellableFSDataOutputStream(
+ awaitResult(mainFileFuture, Duration.Inf),
+ path,
+ awaitResult(checksumFileFuture, Duration.Inf),
+ threadPool
+ )
+ }
+
+ override def open(path: Path): FSDataInputStream = {
+ assert(!isChecksumFile(path), "Cannot directly open a checksum file")
+
+ val checksumInputStreamFuture = Future {
+ try {
+ Some(underlyingFileMgr.open(getChecksumPath(path)))
+ } catch {
+ // In case the client previously had file checksum disabled.
+ // Then previously created files won't have checksum.
+ case _: FileNotFoundException =>
+ logWarning(log"No checksum file found for ${MDC(PATH, path)}, " +
+ log"hence no checksum verification.")
+ None
+ }
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.open(path)
+ }(threadPool)
+
+ val mainStream = awaitResult(mainInputStreamFuture, Duration.Inf)
+ val checksumStream = awaitResult(checksumInputStreamFuture, Duration.Inf)
+
+ checksumStream.map { chkStream =>
+ new ChecksumFSDataInputStream(mainStream, path, chkStream, threadPool)
+ }.getOrElse(mainStream)
+ }
+
+ override def exists(path: Path): Boolean = underlyingFileMgr.exists(path)
+
+ override def delete(path: Path): Unit = {
+ // Allowing directly deleting the checksum file for orphan checksum file scenario
+ if (isChecksumFile(path)) {
+ deleteChecksumFile(path)
+ } else if (allowConcurrentDelete) {
+ // Ideally, we should first try to delete the checksum file
+ // before the main file, to avoid a situation where the main file is deleted but the
+ // checksum file deletion failed. The client might not call delete again if the main file
+ // no longer exists.
+ // But if allowConcurrentDelete is enabled, then we can do it concurrently for perf.
+ // But the client would be responsible for cleaning up potential orphan checksum files
+ // if it happens.
+ val checksumInputStreamFuture = Future {
+ deleteChecksumFile(getChecksumPath(path))
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.delete(path)
+ }(threadPool)
+
+ awaitResult(mainInputStreamFuture, Duration.Inf)
+ awaitResult(checksumInputStreamFuture, Duration.Inf)
+ } else {
+ // First delete the checksum file, then main file
+ deleteChecksumFile(getChecksumPath(path))
+ underlyingFileMgr.delete(path)
+ }
+ }
+
+ private def deleteChecksumFile(checksumPath: Path): Unit = {
+ try {
+ underlyingFileMgr.delete(checksumPath)
+ logDebug(log"Deleted checksum file ${MDC(PATH, checksumPath)}")
+ } catch {
+ case _: FileNotFoundException =>
+ // Ignore if file has already been deleted
+ // or the main file was created initially without checksum
+ logWarning(log"Skipping deletion of checksum file ${MDC(PATH, checksumPath)} " +
+ log"since it does not exist.")
+ }
+ }
+
+ override def isLocal: Boolean = underlyingFileMgr.isLocal
+
+ override def createCheckpointDirectory(): Path = {
+ underlyingFileMgr.createCheckpointDirectory()
+ }
+
+ override def close(): Unit = {
+ threadPool.shutdown()
+ // Wait a bit for it to finish up in case there is any ongoing work
+ // Can consider making this timeout configurable, if needed
+ val timeoutMs = 100
+ if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
+ logWarning(log"Thread pool did not shutdown after ${MDC(TIMEOUT, timeoutMs)} ms")
+ }
+ }
+}
+
+private[streaming] object ChecksumCheckpointFileManager {
+ val CHECKSUM_FILE_SUFFIX = ".crc"
+
+ def awaitResult[T](future: Future[T], atMost: Duration): T = {
+ try {
+ ThreadUtils.awaitResult(future, atMost)
+ } catch {
+ // awaitResult wraps the exception. Unwrap it, and throw the actual error
+ case e: SparkException if e.getMessage.contains("Exception thrown in awaitResult") =>
+ throw e.getCause
+ }
+ }
+
+ def getChecksumPath(mainFilePath: Path): Path = {
+ new Path(mainFilePath.toString + CHECKSUM_FILE_SUFFIX)
+ }
+
+ def isChecksumFile(path: Path): Boolean = {
+ path.getName.endsWith(CHECKSUM_FILE_SUFFIX)
+ }
+}
+
+/** An implementation of [[FSDataInputStream]] that calculates the checksum of the file
+ * that the client is reading (main file) incrementally, while it is being read.
+ * It then does checksum verification on close, to verify that the computed checksum
+ * matches the expected checksum in the checksum file.
+ *
+ * Computing the checksum incrementally and doing the verification after file read is complete
+ * is for better performance, instead of first reading the entire file and doing verification
+ * before the client starts reading the file.
+ *
+ * @param mainStream Input stream for the main file the client wants to read
+ * @param path The path of the main file
+ * @param expectedChecksumStream The input stream for the checksum file
+ * @param threadPool Thread pool to use for concurrently operating on the main and checksum file
+ * */
+class ChecksumFSDataInputStream(
+ private val mainStream: FSDataInputStream,
+ path: Path,
+ private val expectedChecksumStream: FSDataInputStream,
+ private val threadPool: ExecutionContext)
+ extends FSDataInputStream(new CheckedSequentialInputStream(mainStream)) with Logging {
+
+ import ChecksumCheckpointFileManager._
+
+ @volatile private var verified = false
+ @volatile private var closed = false
+
+ override def close(): Unit = {
+ if (!closed) {
+ // We verify the checksum only when the client is done reading.
+ verifyChecksum()
+ closeInternal()
+ }
+ }
+
+ /** This is used to skip checksum verification on close.
+ * Avoid using this, and it is only used for a situation where the file is opened, read,
+ * then closed multiple times, and we want to avoid doing verification each time
+ * and only want to do it once.
+ * */
+ def closeWithoutChecksumVerification(): Unit = {
+ if (!closed) {
+ // Ideally this should be warning, but if a file is doing this frequently
+ // it will cause unnecessary noise in the logs. This can be changed later.
+ logDebug(log"Closing file ${MDC(PATH, path)} without checksum verification")
+ closeInternal()
+ }
+ }
+
+ private def closeInternal(): Unit = {
+ closed = true
+
+ val mainCloseFuture = Future {
+ super.close() // close the main file
+ }(threadPool)
+
+ val checksumCloseFuture = Future {
+ expectedChecksumStream.close() // close the checksum file
+ }(threadPool)
+
+ awaitResult(mainCloseFuture, Duration.Inf)
+ awaitResult(checksumCloseFuture, Duration.Inf)
+ }
+
+ private def verifyChecksum(): Unit = {
+ if (!verified) {
+ // It is possible the file was not read till the end by the reader,
+ // but we need the entire file content for checksum verification.
+ // Hence, we will read the file till the end from where reader stopped.
+ var remainingBytesRead = 0
+ val buffer = new Array[Byte](1024)
+ var bytesRead = 0
+ while ({ bytesRead = super.read(buffer); bytesRead != -1 }) {
+ remainingBytesRead += bytesRead
+ }
+
+ // we are at the end position so that tells us the size
+ val computedFileSize = mainStream.getPos
+
+ if (remainingBytesRead > 0) {
+ // Making this debug log since most of our files are not read exactly to the end
+ // and don't want to cause unnecessary noise in the logs.
+ logDebug(log"File ${MDC(PATH, path)} was not read till the end by reader. " +
+ log"Finished reading the rest of the file for checksum verification. " +
+ log"Remaining bytes read: ${MDC(NUM_BYTES, remainingBytesRead)}, " +
+ log"total size: ${MDC(NUM_BYTES, computedFileSize)}.")
+ }
+
+ // Read the expected checksum from the checksum file
+ val expectedChecksumJson = Source.fromInputStream(
+ expectedChecksumStream, StandardCharsets.UTF_8.name()).mkString
+ val expectedChecksum = Checksum.fromJson(expectedChecksumJson)
+ // Get what we computed while the main file was being read locally
+ // `in` is the CheckedSequentialInputStream we created
+ val computedChecksumValue = in.asInstanceOf[CheckedInputStream].getChecksum.getValue.toInt
+
+ logInfo(log"Verifying checksum for file ${MDC(PATH, path)}, " +
+ log"remainingBytesRead= ${MDC(NUM_BYTES, remainingBytesRead)}. " +
+ log"Computed(checksum= ${MDC(CHECKSUM, computedChecksumValue)}, " +
+ log"fileSize= ${MDC(NUM_BYTES, computedFileSize)})." +
+ log"Checksum file content: ${MDC(CHECKSUM, expectedChecksumJson)}")
+
+ verified = true
+
+ // Compare file size too, in case of collision
+ if (expectedChecksum.value.toInt != computedChecksumValue ||
+ expectedChecksum.mainFileSize != computedFileSize) {
+ throw QueryExecutionErrors.checkpointFileChecksumVerificationFailed(
+ path,
+ expectedSize = expectedChecksum.mainFileSize,
+ expectedChecksum.value,
+ computedSize = computedFileSize,
+ computedChecksumValue.toString)
+ }
+ }
+ }
+}
+
+/** This implements [[CheckedInputStream]] that allows us to compute the checksum as the file
+ * is being read. [[FSDataInputStream]] needs the passed in input stream to implement Seekable
+ * and PositionedReadable. We want the file to be read sequentially only to allow us to correctly
+ * compute the checksum. This is blocking the seekable apis from being used in the underlying stream
+ * */
+private class CheckedSequentialInputStream(data: InputStream)
+ // TODO: Make the checksum algo configurable
Review Comment:
ticket created: https://issues.apache.org/jira/browse/SPARK-52009
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2074771889
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,513 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used and be easily readable.
+ // We can change this to byte array later if we start using algos with large values.
+ value: String,
+ mainFileSize: Long,
+ timestampMs: Long,
+ creator: ChecksumFileCreatorInfo) {
+
+ import Checksum._
+
+ def json(): String = {
+ mapper.writeValueAsString(this)
+ }
+}
+
+object Checksum {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ /** Used to convert between class and JSON. */
+ lazy val mapper = {
+ val _mapper = new ObjectMapper with ClassTagExtensions
+ _mapper.setSerializationInclusion(Include.NON_ABSENT)
+ _mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ _mapper.registerModule(DefaultScalaModule)
+ _mapper
+ }
+
+ def fromJson(json: String): Checksum = {
+ Serialization.read[Checksum](json)
+ }
+}
+
+/** Holds the path of the checksum file and knows the main file path. */
+case class ChecksumFile(path: Path) {
+ import ChecksumCheckpointFileManager._
+ assert(isChecksumFile(path), "path is not a checksum file")
+
+ val mainFilePath = new Path(path.toString.stripSuffix(CHECKSUM_FILE_SUFFIX))
+
+ /** The name of the file without any extensions e.g. my-file.txt.crc returns my-file */
+ val baseName: String = path.getName.split("\\.").head
+}
+
+/**
+ * A [[CheckpointFileManager]] that creates a checksum file for the main file.
+ * This wraps another [[CheckpointFileManager]] and adds checksum functionality on top of it.
+ * Under the hood, when a file is created, it also creates a checksum file with the same name as
+ * the main file but adds a suffix. It returns [[ChecksumCancellableFSDataOutputStream]]
+ * which handles the writing of the main file and checksum file.
+ *
+ * When a file is opened, it returns [[ChecksumFSDataInputStream]], which handles reading
+ * the main file and checksum file and does the checksum verification.
+ *
+ * In order to reduce the impact of reading/writing 2 files instead of 1, it uses a threadpool
+ * to read/write both files concurrently.
+ *
+ * @note
+ * It is able to read files written by other [[CheckpointFileManager]], that don't have checksum.
+ * It automatically deletes the checksum file when the main file is deleted.
+ * If you delete the main file with a different type of manager, then the checksum file will be
+ * left behind (i.e. orphan checksum file), since they don't know about it. It would be your
+ * responsibility to delete the orphan checksum files.
+ *
+ * @param underlyingFileMgr The file manager to use under the hood
+ * @param allowConcurrentDelete If true, allows deleting the main and checksum file concurrently.
+ * This is a perf optimization, but can potentially lead to
+ * orphan checksum files. If using this, it is your responsibility
+ * to clean up the potential orphan checksum files.
+ * @param numThreads This is the number of threads to use for the thread pool, for reading/writing
+ * files. To avoid blocking, if the file manager instance is being used by a
+ * single thread, then you can set this to 2 (one thread for main file, another
+ * for checksum file).
+ * If file manager is shared by multiple threads, you can set it to
+ * number of threads using file manager * 2.
+ * Setting this differently can lead to file operation being blocked waiting for
+ * a free thread.
+ */
+class ChecksumCheckpointFileManager(
+ private val underlyingFileMgr: CheckpointFileManager,
+ val allowConcurrentDelete: Boolean = false,
+ val numThreads: Int)
+ extends CheckpointFileManager with Logging {
+ assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
+ "and another for the checksum file")
+
+ import ChecksumCheckpointFileManager._
+
+ // This allows us to concurrently read/write the main file and checksum file
+ private val threadPool = ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+
+ override def list(path: Path, filter: PathFilter): Array[FileStatus] = {
+ underlyingFileMgr.list(path, filter)
+ }
+
+ override def mkdirs(path: Path): Unit = {
+ underlyingFileMgr.mkdirs(path)
+ }
+
+ override def createAtomic(path: Path,
+ overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
+ assert(!isChecksumFile(path), "Cannot directly create a checksum file")
+
+ val mainFileFuture = Future {
+ underlyingFileMgr.createAtomic(path, overwriteIfPossible)
+ }(threadPool)
+
+ val checksumFileFuture = Future {
+ underlyingFileMgr.createAtomic(getChecksumPath(path), overwriteIfPossible)
+ }(threadPool)
+
+ new ChecksumCancellableFSDataOutputStream(
+ awaitResult(mainFileFuture, Duration.Inf),
+ path,
+ awaitResult(checksumFileFuture, Duration.Inf),
+ threadPool
+ )
+ }
+
+ override def open(path: Path): FSDataInputStream = {
+ assert(!isChecksumFile(path), "Cannot directly open a checksum file")
+
+ val checksumInputStreamFuture = Future {
+ try {
+ Some(underlyingFileMgr.open(getChecksumPath(path)))
+ } catch {
+ // In case the client previously had file checksum disabled.
+ // Then previously created files won't have checksum.
+ case _: FileNotFoundException =>
+ logWarning(log"No checksum file found for ${MDC(PATH, path)}, " +
+ log"hence no checksum verification.")
+ None
+ }
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.open(path)
+ }(threadPool)
+
+ val mainStream = awaitResult(mainInputStreamFuture, Duration.Inf)
+ val checksumStream = awaitResult(checksumInputStreamFuture, Duration.Inf)
+
+ checksumStream.map { chkStream =>
+ new ChecksumFSDataInputStream(mainStream, path, chkStream, threadPool)
+ }.getOrElse(mainStream)
+ }
+
+ override def exists(path: Path): Boolean = underlyingFileMgr.exists(path)
+
+ override def delete(path: Path): Unit = {
+ // Allowing directly deleting the checksum file for orphan checksum file scenario
+ if (isChecksumFile(path)) {
+ deleteChecksumFile(path)
+ } else if (allowConcurrentDelete) {
+ // Ideally, we should first try to delete the checksum file
+ // before the main file, to avoid a situation where the main file is deleted but the
+ // checksum file deletion failed. The client might not call delete again if the main file
+ // no longer exists.
+ // But if allowConcurrentDelete is enabled, then we can do it concurrently for perf.
+ // But the client would be responsible for cleaning up potential orphan checksum files
+ // if it happens.
+ val checksumInputStreamFuture = Future {
+ deleteChecksumFile(getChecksumPath(path))
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.delete(path)
+ }(threadPool)
+
+ awaitResult(mainInputStreamFuture, Duration.Inf)
+ awaitResult(checksumInputStreamFuture, Duration.Inf)
+ } else {
+ // First delete the checksum file, then main file
+ deleteChecksumFile(getChecksumPath(path))
+ underlyingFileMgr.delete(path)
+ }
+ }
+
+ private def deleteChecksumFile(checksumPath: Path): Unit = {
+ try {
+ underlyingFileMgr.delete(checksumPath)
+ logDebug(log"Deleted checksum file ${MDC(PATH, checksumPath)}")
+ } catch {
+ case _: FileNotFoundException =>
+ // Ignore if file has already been deleted
+ // or the main file was created initially without checksum
+ logWarning(log"Skipping deletion of checksum file ${MDC(PATH, checksumPath)} " +
+ log"since it does not exist.")
+ }
+ }
+
+ override def isLocal: Boolean = underlyingFileMgr.isLocal
+
+ override def createCheckpointDirectory(): Path = {
+ underlyingFileMgr.createCheckpointDirectory()
+ }
+
+ override def close(): Unit = {
+ threadPool.shutdown()
+ // Wait a bit for it to finish up in case there is any ongoing work
+ // Can consider making this timeout configurable, if needed
+ val timeoutMs = 100
+ if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
Review Comment:
done
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072008889
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,513 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used and be easily readable.
+ // We can change this to byte array later if we start using algos with large values.
+ value: String,
+ mainFileSize: Long,
+ timestampMs: Long,
+ creator: ChecksumFileCreatorInfo) {
+
+ import Checksum._
+
+ def json(): String = {
+ mapper.writeValueAsString(this)
+ }
+}
+
+object Checksum {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ /** Used to convert between class and JSON. */
+ lazy val mapper = {
+ val _mapper = new ObjectMapper with ClassTagExtensions
+ _mapper.setSerializationInclusion(Include.NON_ABSENT)
+ _mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ _mapper.registerModule(DefaultScalaModule)
+ _mapper
+ }
+
+ def fromJson(json: String): Checksum = {
+ Serialization.read[Checksum](json)
+ }
+}
+
+/** Holds the path of the checksum file and knows the main file path. */
+case class ChecksumFile(path: Path) {
+ import ChecksumCheckpointFileManager._
+ assert(isChecksumFile(path), "path is not a checksum file")
+
+ val mainFilePath = new Path(path.toString.stripSuffix(CHECKSUM_FILE_SUFFIX))
+
+ /** The name of the file without any extensions e.g. my-file.txt.crc returns my-file */
+ val baseName: String = path.getName.split("\\.").head
+}
+
+/**
+ * A [[CheckpointFileManager]] that creates a checksum file for the main file.
+ * This wraps another [[CheckpointFileManager]] and adds checksum functionality on top of it.
+ * Under the hood, when a file is created, it also creates a checksum file with the same name as
+ * the main file but adds a suffix. It returns [[ChecksumCancellableFSDataOutputStream]]
+ * which handles the writing of the main file and checksum file.
+ *
+ * When a file is opened, it returns [[ChecksumFSDataInputStream]], which handles reading
+ * the main file and checksum file and does the checksum verification.
+ *
+ * In order to reduce the impact of reading/writing 2 files instead of 1, it uses a threadpool
+ * to read/write both files concurrently.
+ *
+ * @note
+ * It is able to read files written by other [[CheckpointFileManager]], that don't have checksum.
+ * It automatically deletes the checksum file when the main file is deleted.
+ * If you delete the main file with a different type of manager, then the checksum file will be
+ * left behind (i.e. orphan checksum file), since they don't know about it. It would be your
+ * responsibility to delete the orphan checksum files.
+ *
+ * @param underlyingFileMgr The file manager to use under the hood
+ * @param allowConcurrentDelete If true, allows deleting the main and checksum file concurrently.
+ * This is a perf optimization, but can potentially lead to
+ * orphan checksum files. If using this, it is your responsibility
+ * to clean up the potential orphan checksum files.
+ * @param numThreads This is the number of threads to use for the thread pool, for reading/writing
+ * files. To avoid blocking, if the file manager instance is being used by a
+ * single thread, then you can set this to 2 (one thread for main file, another
+ * for checksum file).
+ * If file manager is shared by multiple threads, you can set it to
+ * number of threads using file manager * 2.
+ * Setting this differently can lead to file operation being blocked waiting for
+ * a free thread.
+ */
+class ChecksumCheckpointFileManager(
+ private val underlyingFileMgr: CheckpointFileManager,
+ val allowConcurrentDelete: Boolean = false,
+ val numThreads: Int)
+ extends CheckpointFileManager with Logging {
+ assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
+ "and another for the checksum file")
+
+ import ChecksumCheckpointFileManager._
+
+ // This allows us to concurrently read/write the main file and checksum file
+ private val threadPool = ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+
+ override def list(path: Path, filter: PathFilter): Array[FileStatus] = {
+ underlyingFileMgr.list(path, filter)
+ }
+
+ override def mkdirs(path: Path): Unit = {
+ underlyingFileMgr.mkdirs(path)
+ }
+
+ override def createAtomic(path: Path,
+ overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
+ assert(!isChecksumFile(path), "Cannot directly create a checksum file")
+
+ val mainFileFuture = Future {
+ underlyingFileMgr.createAtomic(path, overwriteIfPossible)
+ }(threadPool)
+
+ val checksumFileFuture = Future {
+ underlyingFileMgr.createAtomic(getChecksumPath(path), overwriteIfPossible)
+ }(threadPool)
+
+ new ChecksumCancellableFSDataOutputStream(
+ awaitResult(mainFileFuture, Duration.Inf),
+ path,
+ awaitResult(checksumFileFuture, Duration.Inf),
+ threadPool
+ )
+ }
+
+ override def open(path: Path): FSDataInputStream = {
+ assert(!isChecksumFile(path), "Cannot directly open a checksum file")
+
+ val checksumInputStreamFuture = Future {
+ try {
+ Some(underlyingFileMgr.open(getChecksumPath(path)))
+ } catch {
+ // In case the client previously had file checksum disabled.
+ // Then previously created files won't have checksum.
+ case _: FileNotFoundException =>
+ logWarning(log"No checksum file found for ${MDC(PATH, path)}, " +
+ log"hence no checksum verification.")
+ None
+ }
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.open(path)
+ }(threadPool)
+
+ val mainStream = awaitResult(mainInputStreamFuture, Duration.Inf)
+ val checksumStream = awaitResult(checksumInputStreamFuture, Duration.Inf)
+
+ checksumStream.map { chkStream =>
+ new ChecksumFSDataInputStream(mainStream, path, chkStream, threadPool)
+ }.getOrElse(mainStream)
+ }
+
+ override def exists(path: Path): Boolean = underlyingFileMgr.exists(path)
+
+ override def delete(path: Path): Unit = {
+ // Allowing directly deleting the checksum file for orphan checksum file scenario
+ if (isChecksumFile(path)) {
+ deleteChecksumFile(path)
+ } else if (allowConcurrentDelete) {
+ // Ideally, we should first try to delete the checksum file
+ // before the main file, to avoid a situation where the main file is deleted but the
+ // checksum file deletion failed. The client might not call delete again if the main file
+ // no longer exists.
+ // But if allowConcurrentDelete is enabled, then we can do it concurrently for perf.
+ // But the client would be responsible for cleaning up potential orphan checksum files
+ // if it happens.
+ val checksumInputStreamFuture = Future {
+ deleteChecksumFile(getChecksumPath(path))
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.delete(path)
+ }(threadPool)
+
+ awaitResult(mainInputStreamFuture, Duration.Inf)
+ awaitResult(checksumInputStreamFuture, Duration.Inf)
+ } else {
+ // First delete the checksum file, then main file
+ deleteChecksumFile(getChecksumPath(path))
+ underlyingFileMgr.delete(path)
+ }
+ }
+
+ private def deleteChecksumFile(checksumPath: Path): Unit = {
+ try {
+ underlyingFileMgr.delete(checksumPath)
+ logDebug(log"Deleted checksum file ${MDC(PATH, checksumPath)}")
+ } catch {
+ case _: FileNotFoundException =>
+ // Ignore if file has already been deleted
+ // or the main file was created initially without checksum
+ logWarning(log"Skipping deletion of checksum file ${MDC(PATH, checksumPath)} " +
+ log"since it does not exist.")
+ }
+ }
+
+ override def isLocal: Boolean = underlyingFileMgr.isLocal
+
+ override def createCheckpointDirectory(): Path = {
+ underlyingFileMgr.createCheckpointDirectory()
+ }
+
+ override def close(): Unit = {
+ threadPool.shutdown()
+ // Wait a bit for it to finish up in case there is any ongoing work
+ // Can consider making this timeout configurable, if needed
+ val timeoutMs = 100
+ if (!threadPool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
+ logWarning(log"Thread pool did not shutdown after ${MDC(TIMEOUT, timeoutMs)} ms")
+ }
+ }
+}
+
+private[streaming] object ChecksumCheckpointFileManager {
+ val CHECKSUM_FILE_SUFFIX = ".crc"
+
+ def awaitResult[T](future: Future[T], atMost: Duration): T = {
+ try {
+ ThreadUtils.awaitResult(future, atMost)
+ } catch {
+ // awaitResult wraps the exception. Unwrap it, and throw the actual error
+ case e: SparkException if e.getMessage.contains("Exception thrown in awaitResult") =>
+ throw e.getCause
+ }
+ }
+
+ def getChecksumPath(mainFilePath: Path): Path = {
+ new Path(mainFilePath.toString + CHECKSUM_FILE_SUFFIX)
+ }
+
+ def isChecksumFile(path: Path): Boolean = {
+ path.getName.endsWith(CHECKSUM_FILE_SUFFIX)
+ }
+}
+
+/** An implementation of [[FSDataInputStream]] that calculates the checksum of the file
+ * that the client is reading (main file) incrementally, while it is being read.
+ * It then does checksum verification on close, to verify that the computed checksum
+ * matches the expected checksum in the checksum file.
+ *
+ * Computing the checksum incrementally and doing the verification after file read is complete
+ * is for better performance, instead of first reading the entire file and doing verification
+ * before the client starts reading the file.
+ *
+ * @param mainStream Input stream for the main file the client wants to read
+ * @param path The path of the main file
+ * @param expectedChecksumStream The input stream for the checksum file
+ * @param threadPool Thread pool to use for concurrently operating on the main and checksum file
+ * */
+class ChecksumFSDataInputStream(
+ private val mainStream: FSDataInputStream,
+ path: Path,
+ private val expectedChecksumStream: FSDataInputStream,
+ private val threadPool: ExecutionContext)
+ extends FSDataInputStream(new CheckedSequentialInputStream(mainStream)) with Logging {
+
+ import ChecksumCheckpointFileManager._
+
+ @volatile private var verified = false
+ @volatile private var closed = false
+
+ override def close(): Unit = {
+ if (!closed) {
+ // We verify the checksum only when the client is done reading.
+ verifyChecksum()
+ closeInternal()
+ }
+ }
+
+ /** This is used to skip checksum verification on close.
+ * Avoid using this, and it is only used for a situation where the file is opened, read,
+ * then closed multiple times, and we want to avoid doing verification each time
+ * and only want to do it once.
+ * */
+ def closeWithoutChecksumVerification(): Unit = {
+ if (!closed) {
+ // Ideally this should be warning, but if a file is doing this frequently
+ // it will cause unnecessary noise in the logs. This can be changed later.
+ logDebug(log"Closing file ${MDC(PATH, path)} without checksum verification")
+ closeInternal()
+ }
+ }
+
+ private def closeInternal(): Unit = {
+ closed = true
+
+ val mainCloseFuture = Future {
+ super.close() // close the main file
+ }(threadPool)
+
+ val checksumCloseFuture = Future {
+ expectedChecksumStream.close() // close the checksum file
+ }(threadPool)
+
+ awaitResult(mainCloseFuture, Duration.Inf)
+ awaitResult(checksumCloseFuture, Duration.Inf)
+ }
+
+ private def verifyChecksum(): Unit = {
+ if (!verified) {
+ // It is possible the file was not read till the end by the reader,
+ // but we need the entire file content for checksum verification.
+ // Hence, we will read the file till the end from where reader stopped.
+ var remainingBytesRead = 0
+ val buffer = new Array[Byte](1024)
+ var bytesRead = 0
+ while ({ bytesRead = super.read(buffer); bytesRead != -1 }) {
+ remainingBytesRead += bytesRead
+ }
+
+ // we are at the end position so that tells us the size
+ val computedFileSize = mainStream.getPos
+
+ if (remainingBytesRead > 0) {
+ // Making this debug log since most of our files are not read exactly to the end
+ // and don't want to cause unnecessary noise in the logs.
+ logDebug(log"File ${MDC(PATH, path)} was not read till the end by reader. " +
+ log"Finished reading the rest of the file for checksum verification. " +
+ log"Remaining bytes read: ${MDC(NUM_BYTES, remainingBytesRead)}, " +
+ log"total size: ${MDC(NUM_BYTES, computedFileSize)}.")
+ }
+
+ // Read the expected checksum from the checksum file
+ val expectedChecksumJson = Source.fromInputStream(
+ expectedChecksumStream, StandardCharsets.UTF_8.name()).mkString
+ val expectedChecksum = Checksum.fromJson(expectedChecksumJson)
+ // Get what we computed while the main file was being read locally
+ // `in` is the CheckedSequentialInputStream we created
+ val computedChecksumValue = in.asInstanceOf[CheckedInputStream].getChecksum.getValue.toInt
+
+ logInfo(log"Verifying checksum for file ${MDC(PATH, path)}, " +
+ log"remainingBytesRead= ${MDC(NUM_BYTES, remainingBytesRead)}. " +
+ log"Computed(checksum= ${MDC(CHECKSUM, computedChecksumValue)}, " +
+ log"fileSize= ${MDC(NUM_BYTES, computedFileSize)})." +
+ log"Checksum file content: ${MDC(CHECKSUM, expectedChecksumJson)}")
+
+ verified = true
+
+ // Compare file size too, in case of collision
+ if (expectedChecksum.value.toInt != computedChecksumValue ||
+ expectedChecksum.mainFileSize != computedFileSize) {
+ throw QueryExecutionErrors.checkpointFileChecksumVerificationFailed(
+ path,
+ expectedSize = expectedChecksum.mainFileSize,
+ expectedChecksum.value,
+ computedSize = computedFileSize,
+ computedChecksumValue.toString)
+ }
+ }
+ }
+}
+
+/** This implements [[CheckedInputStream]] that allows us to compute the checksum as the file
+ * is being read. [[FSDataInputStream]] needs the passed in input stream to implement Seekable
+ * and PositionedReadable. We want the file to be read sequentially only to allow us to correctly
+ * compute the checksum. This is blocking the seekable apis from being used in the underlying stream
+ * */
+private class CheckedSequentialInputStream(data: InputStream)
+ // TODO: Make the checksum algo configurable
Review Comment:
lets add it in this change itself ? if you want to do this as a follow up - lets create the SPARK JIRA for this and link it here ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51972][SS] State Store file integrity verification using checksum [spark]
Posted by "micheal-o (via GitHub)" <gi...@apache.org>.
micheal-o commented on code in PR #50773:
URL: https://github.com/apache/spark/pull/50773#discussion_r2072016084
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManager.scala:
##########
@@ -0,0 +1,513 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, InputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.TimeUnit
+import java.util.zip.{CheckedInputStream, CheckedOutputStream, CRC32C}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration.Duration
+import scala.io.Source
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.hadoop.fs._
+import org.json4s.{Formats, NoTypeHints}
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.SparkException
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys.{CHECKSUM, NUM_BYTES, PATH, TIMEOUT}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager.CancellableFSDataOutputStream
+import org.apache.spark.util.ThreadUtils
+
+/** Information about the creator of the checksum file. Useful for debugging */
+case class ChecksumFileCreatorInfo(
+ executorId: String,
+ taskInfo: String)
+
+object ChecksumFileCreatorInfo {
+ def apply(): ChecksumFileCreatorInfo = {
+ val executorId = Option(SparkEnv.get).map(_.executorId).getOrElse("")
+ val taskInfo = Option(TaskContext.get()).map(tc =>
+ s"Task= ${tc.partitionId()}.${tc.attemptNumber()}, " +
+ s"Stage= ${tc.stageId()}.${tc.stageAttemptNumber()}").getOrElse("")
+ new ChecksumFileCreatorInfo(executorId, taskInfo)
+ }
+}
+
+/** This is the content of the checksum file.
+ * Holds the checksum value and additional information */
+case class Checksum(
+ algorithm: String,
+ // Making this a string to be agnostic of algorithm used and be easily readable.
+ // We can change this to byte array later if we start using algos with large values.
+ value: String,
+ mainFileSize: Long,
+ timestampMs: Long,
+ creator: ChecksumFileCreatorInfo) {
+
+ import Checksum._
+
+ def json(): String = {
+ mapper.writeValueAsString(this)
+ }
+}
+
+object Checksum {
+ implicit val format: Formats = Serialization.formats(NoTypeHints)
+
+ /** Used to convert between class and JSON. */
+ lazy val mapper = {
+ val _mapper = new ObjectMapper with ClassTagExtensions
+ _mapper.setSerializationInclusion(Include.NON_ABSENT)
+ _mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ _mapper.registerModule(DefaultScalaModule)
+ _mapper
+ }
+
+ def fromJson(json: String): Checksum = {
+ Serialization.read[Checksum](json)
+ }
+}
+
+/** Holds the path of the checksum file and knows the main file path. */
+case class ChecksumFile(path: Path) {
+ import ChecksumCheckpointFileManager._
+ assert(isChecksumFile(path), "path is not a checksum file")
+
+ val mainFilePath = new Path(path.toString.stripSuffix(CHECKSUM_FILE_SUFFIX))
+
+ /** The name of the file without any extensions e.g. my-file.txt.crc returns my-file */
+ val baseName: String = path.getName.split("\\.").head
+}
+
+/**
+ * A [[CheckpointFileManager]] that creates a checksum file for the main file.
+ * This wraps another [[CheckpointFileManager]] and adds checksum functionality on top of it.
+ * Under the hood, when a file is created, it also creates a checksum file with the same name as
+ * the main file but adds a suffix. It returns [[ChecksumCancellableFSDataOutputStream]]
+ * which handles the writing of the main file and checksum file.
+ *
+ * When a file is opened, it returns [[ChecksumFSDataInputStream]], which handles reading
+ * the main file and checksum file and does the checksum verification.
+ *
+ * In order to reduce the impact of reading/writing 2 files instead of 1, it uses a threadpool
+ * to read/write both files concurrently.
+ *
+ * @note
+ * It is able to read files written by other [[CheckpointFileManager]], that don't have checksum.
+ * It automatically deletes the checksum file when the main file is deleted.
+ * If you delete the main file with a different type of manager, then the checksum file will be
+ * left behind (i.e. orphan checksum file), since they don't know about it. It would be your
+ * responsibility to delete the orphan checksum files.
+ *
+ * @param underlyingFileMgr The file manager to use under the hood
+ * @param allowConcurrentDelete If true, allows deleting the main and checksum file concurrently.
+ * This is a perf optimization, but can potentially lead to
+ * orphan checksum files. If using this, it is your responsibility
+ * to clean up the potential orphan checksum files.
+ * @param numThreads This is the number of threads to use for the thread pool, for reading/writing
+ * files. To avoid blocking, if the file manager instance is being used by a
+ * single thread, then you can set this to 2 (one thread for main file, another
+ * for checksum file).
+ * If file manager is shared by multiple threads, you can set it to
+ * number of threads using file manager * 2.
+ * Setting this differently can lead to file operation being blocked waiting for
+ * a free thread.
+ */
+class ChecksumCheckpointFileManager(
+ private val underlyingFileMgr: CheckpointFileManager,
+ val allowConcurrentDelete: Boolean = false,
+ val numThreads: Int)
+ extends CheckpointFileManager with Logging {
+ assert(numThreads % 2 == 0, "numThreads must be a multiple of 2, we need 1 for the main file" +
+ "and another for the checksum file")
+
+ import ChecksumCheckpointFileManager._
+
+ // This allows us to concurrently read/write the main file and checksum file
+ private val threadPool = ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonFixedThreadPool(numThreads, s"${this.getClass.getSimpleName}-Thread"))
+
+ override def list(path: Path, filter: PathFilter): Array[FileStatus] = {
+ underlyingFileMgr.list(path, filter)
+ }
+
+ override def mkdirs(path: Path): Unit = {
+ underlyingFileMgr.mkdirs(path)
+ }
+
+ override def createAtomic(path: Path,
+ overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
+ assert(!isChecksumFile(path), "Cannot directly create a checksum file")
+
+ val mainFileFuture = Future {
+ underlyingFileMgr.createAtomic(path, overwriteIfPossible)
+ }(threadPool)
+
+ val checksumFileFuture = Future {
+ underlyingFileMgr.createAtomic(getChecksumPath(path), overwriteIfPossible)
+ }(threadPool)
+
+ new ChecksumCancellableFSDataOutputStream(
+ awaitResult(mainFileFuture, Duration.Inf),
+ path,
+ awaitResult(checksumFileFuture, Duration.Inf),
+ threadPool
+ )
+ }
+
+ override def open(path: Path): FSDataInputStream = {
+ assert(!isChecksumFile(path), "Cannot directly open a checksum file")
+
+ val checksumInputStreamFuture = Future {
+ try {
+ Some(underlyingFileMgr.open(getChecksumPath(path)))
+ } catch {
+ // In case the client previously had file checksum disabled.
+ // Then previously created files won't have checksum.
+ case _: FileNotFoundException =>
+ logWarning(log"No checksum file found for ${MDC(PATH, path)}, " +
+ log"hence no checksum verification.")
+ None
+ }
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.open(path)
+ }(threadPool)
+
+ val mainStream = awaitResult(mainInputStreamFuture, Duration.Inf)
+ val checksumStream = awaitResult(checksumInputStreamFuture, Duration.Inf)
+
+ checksumStream.map { chkStream =>
+ new ChecksumFSDataInputStream(mainStream, path, chkStream, threadPool)
+ }.getOrElse(mainStream)
+ }
+
+ override def exists(path: Path): Boolean = underlyingFileMgr.exists(path)
+
+ override def delete(path: Path): Unit = {
+ // Allowing directly deleting the checksum file for orphan checksum file scenario
+ if (isChecksumFile(path)) {
+ deleteChecksumFile(path)
+ } else if (allowConcurrentDelete) {
+ // Ideally, we should first try to delete the checksum file
+ // before the main file, to avoid a situation where the main file is deleted but the
+ // checksum file deletion failed. The client might not call delete again if the main file
+ // no longer exists.
+ // But if allowConcurrentDelete is enabled, then we can do it concurrently for perf.
+ // But the client would be responsible for cleaning up potential orphan checksum files
+ // if it happens.
+ val checksumInputStreamFuture = Future {
+ deleteChecksumFile(getChecksumPath(path))
+ }(threadPool)
+
+ val mainInputStreamFuture = Future {
+ underlyingFileMgr.delete(path)
+ }(threadPool)
+
+ awaitResult(mainInputStreamFuture, Duration.Inf)
+ awaitResult(checksumInputStreamFuture, Duration.Inf)
+ } else {
+ // First delete the checksum file, then main file
+ deleteChecksumFile(getChecksumPath(path))
+ underlyingFileMgr.delete(path)
+ }
+ }
+
+ private def deleteChecksumFile(checksumPath: Path): Unit = {
+ try {
+ underlyingFileMgr.delete(checksumPath)
+ logDebug(log"Deleted checksum file ${MDC(PATH, checksumPath)}")
+ } catch {
+ case _: FileNotFoundException =>
+ // Ignore if file has already been deleted
+ // or the main file was created initially without checksum
+ logWarning(log"Skipping deletion of checksum file ${MDC(PATH, checksumPath)} " +
Review Comment:
Actually, the underlying delete call doesn't actually throw error if file not found (It is part of requirement of checkpointfm api), so I will just remove this catch anyway.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org