This is an automated email from the ASF dual-hosted git repository.
ashrigondekar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 811482a79cbc [SPARK-54290][SS] Skip checksum creation if path already
exists without checksum
811482a79cbc is described below
commit 811482a79cbc9b26c1502cbaed114340e9a82379
Author: Dylan Wong <[email protected]>
AuthorDate: Tue Nov 11 10:20:29 2025 -0800
[SPARK-54290][SS] Skip checksum creation if path already exists without
checksum
### What changes were proposed in this pull request?
This PR modifies ChecksumCheckpointFileManager to fall back to the
underlying CheckpointFileManager when the target path exists but the checksum
file does not. This prevents failures in checkpoint recovery scenarios where
checksum validation cannot be performed.
### Why are the changes needed?
Consider the case using STATE_STORE_CHECKPOINT_FORMAT_VERSION = 1 when a
batch fails but state files are written. If on the next run, we try to upload
both a new state file and a file checksum, the file could fail to be uploaded
but the file checksum is uploaded successfully. This would lead to a situation
where the old file could be loaded and compared with the new file checksum,
which would fail the checksum verification. This issue does not happen when
STATE_STORE_CHECKPOINT_FORMAT [...]
### Does this PR introduce _any_ user-facing change?
### How was this patch tested?
- Added unit tests in ChecksumCheckpointFileManagerSuite to verify fallback
behavior
- Added a failure injection test in RocksDBCheckpointFailureInjectionSuite
to simulate how this error is caused
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #52985 from dylanwong250/SPARK-54290.
Authored-by: Dylan Wong <[email protected]>
Signed-off-by: Anish Shrigondekar <[email protected]>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 13 ++
.../ChecksumCheckpointFileManager.scala | 30 ++-
.../state/HDFSBackedStateStoreProvider.scala | 4 +-
.../sql/execution/streaming/state/RocksDB.scala | 15 +-
.../streaming/state/RocksDBFileManager.scala | 5 +-
.../execution/streaming/state/StateStoreConf.scala | 14 ++
.../ChecksumCheckpointFileManagerSuite.scala | 80 +++++--
.../FailureInjectionCheckpointFileManager.scala | 6 +-
.../RocksDBCheckpointFailureInjectionSuite.scala | 239 ++++++++++++++++++++-
9 files changed, 380 insertions(+), 26 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 36ded2bd7b63..8fbed6acc7d0 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -3515,6 +3515,16 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val
STREAMING_CHECKPOINT_FILE_CHECKSUM_SKIP_CREATION_IF_FILE_MISSING_CHECKSUM =
+
buildConf("spark.sql.streaming.checkpoint.fileChecksum.skipCreationIfFileMissingChecksum")
+ .internal()
+ .doc("When true, if a microbatch is retried, if a file already exists
but its checksum " +
+ "file does not exist, the file checksum will not be created. This is
useful for " +
+ "compatibility with files created before file checksums were enabled.")
+ .version("4.2.0")
+ .booleanConf
+ .createWithDefault(true)
+
val PARALLEL_FILE_LISTING_IN_STATS_COMPUTATION =
buildConf("spark.sql.statistics.parallelFileListingInStatsComputation.enabled")
.internal()
@@ -6855,6 +6865,9 @@ class SQLConf extends Serializable with Logging with
SqlApiConf {
def checkpointFileChecksumEnabled: Boolean =
getConf(STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED)
+ def checkpointFileChecksumSkipCreationIfFileMissingChecksum: Boolean =
+
getConf(STREAMING_CHECKPOINT_FILE_CHECKSUM_SKIP_CREATION_IF_FILE_MISSING_CHECKSUM)
+
def isUnsupportedOperationCheckEnabled: Boolean =
getConf(UNSUPPORTED_OPERATION_CHECK_ENABLED)
def useDeprecatedKafkaOffsetFetching: Boolean =
getConf(USE_DEPRECATED_KAFKA_OFFSET_FETCHING)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala
index 7f801392c2f4..637d11ad890b 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/checkpointing/ChecksumCheckpointFileManager.scala
@@ -133,11 +133,21 @@ case class ChecksumFile(path: Path) {
* number of threads using file manager * 2.
* Setting this differently can lead to file operation being
blocked waiting for
* a free thread.
+ * @param skipCreationIfFileMissingChecksum (ES-1629547): If true, when a file
already exists
+ * but its checksum file does not exist, fall back to using
the underlying
+ * file manager directly instead of creating with checksum.
This is useful
+ * for compatibility with files created before checksums
were enabled. Consider
+ * the case when a batch fails but state files are written.
If on the next run,
+ * we try to upload both a new file and a checksum file, the
file could fail to be
+ * uploaded but the checksum file is uploaded successfully.
This would lead to a
+ * situation where the old file could be loaded and compared
with the new file
+ * checksum, which would fail the checksum verification.
*/
class ChecksumCheckpointFileManager(
private val underlyingFileMgr: CheckpointFileManager,
val allowConcurrentDelete: Boolean = false,
- val numThreads: Int)
+ val numThreads: Int,
+ val skipCreationIfFileMissingChecksum: Boolean)
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")
@@ -160,9 +170,18 @@ class ChecksumCheckpointFileManager(
underlyingFileMgr.mkdirs(path)
}
+ private def shouldSkipChecksumCreation(path: Path): Boolean = {
+ skipCreationIfFileMissingChecksum &&
+ underlyingFileMgr.exists(path) &&
!underlyingFileMgr.exists(getChecksumPath(path))
+ }
+
override def createAtomic(path: Path,
overwriteIfPossible: Boolean): CancellableFSDataOutputStream = {
- createWithChecksum(path, underlyingFileMgr.createAtomic(_,
overwriteIfPossible))
+ if (shouldSkipChecksumCreation(path)) {
+ underlyingFileMgr.createAtomic(path, overwriteIfPossible)
+ } else {
+ createWithChecksum(path, underlyingFileMgr.createAtomic(_,
overwriteIfPossible))
+ }
}
private def createWithChecksum(path: Path,
@@ -327,8 +346,11 @@ class ChecksumFSDataInputStream(
override def close(): Unit = {
if (!closed) {
// We verify the checksum only when the client is done reading.
- verifyChecksum()
- closeInternal()
+ try {
+ verifyChecksum()
+ } finally {
+ closeInternal()
+ }
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
index a0ace7976edd..3bbdc1fa6785 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala
@@ -487,7 +487,9 @@ private[sql] class HDFSBackedStateStoreProvider extends
StateStoreProvider with
// (one for main file and another for checksum file).
// Since this fm is used by both query task and maintenance thread,
// then we need 2 * 2 = 4 threads.
- numThreads = 4)
+ numThreads = 4,
+ skipCreationIfFileMissingChecksum =
+ storeConf.checkpointFileChecksumSkipCreationIfFileMissingChecksum)
} else {
mgr
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
index b1c9dee5a459..b8582d484538 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
@@ -150,20 +150,29 @@ class RocksDB(
localTempDir: File,
hadoopConf: Configuration,
codecName: String,
- loggingId: String): RocksDBFileManager = {
+ loggingId: String,
+ storeConf: StateStoreConf): RocksDBFileManager = {
new RocksDBFileManager(
dfsRootDir,
localTempDir,
hadoopConf,
codecName,
loggingId = loggingId,
+ storeConf,
fileChecksumEnabled = conf.fileChecksumEnabled,
fileChecksumThreadPoolSize = fileChecksumThreadPoolSize
)
}
- private[spark] val fileManager = createFileManager(dfsRootDir,
createTempDir("fileManager"),
- hadoopConf, conf.compressionCodec, loggingId = loggingId)
+ private[spark] val fileManager = createFileManager(
+ dfsRootDir,
+ createTempDir("fileManager"),
+ hadoopConf,
+ conf.compressionCodec,
+ loggingId = loggingId,
+ storeConf = conf.stateStoreConf
+ )
+
private val byteArrayPair = new ByteArrayPair()
private val commitLatencyMs = new mutable.HashMap[String, Long]()
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
index 92fa5d0350fa..f67d80679d51 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
@@ -132,6 +132,7 @@ class RocksDBFileManager(
hadoopConf: Configuration,
codecName: String = CompressionCodec.ZSTD,
loggingId: String = "",
+ storeConf: StateStoreConf = StateStoreConf.empty,
fileChecksumEnabled: Boolean = false,
fileChecksumThreadPoolSize: Option[Int] = None)
extends Logging {
@@ -149,7 +150,9 @@ class RocksDBFileManager(
mgr,
// Allowing this for perf, since we do orphan checksum file cleanup in
maintenance anyway
allowConcurrentDelete = true,
- numThreads = fileChecksumThreadPoolSize.get)
+ numThreads = fileChecksumThreadPoolSize.get,
+ skipCreationIfFileMissingChecksum
+ = storeConf.checkpointFileChecksumSkipCreationIfFileMissingChecksum)
} else {
mgr
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala
index ebb212512ccb..3991f8d93f2c 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreConf.scala
@@ -125,6 +125,20 @@ class StateStoreConf(
val enableStateStoreCheckpointIds =
StatefulOperatorStateInfo.enableStateStoreCheckpointIds(sqlConf)
+ /**
+ * Whether to skip checksum creation if file missing checksum.
+ *
+ * Consider the case using STATE_STORE_CHECKPOINT_FORMAT_VERSION = 1 when a
batch fails but state
+ * files are written. If on the next run, we try to upload both a new state
file and a file
+ * checksum, the file could fail to be uploaded but the file checksum is
uploaded successfully.
+ * This would lead to a situation where the old file could be loaded and
compared with the new
+ * file checksum, which would fail the checksum verification. This issue
does not happen when
+ * STATE_STORE_CHECKPOINT_FORMAT_VERSION = 2 since each batch run unique ids
will be created.
+ */
+ val checkpointFileChecksumSkipCreationIfFileMissingChecksum: Boolean =
+ sqlConf.checkpointFileChecksumSkipCreationIfFileMissingChecksum &&
+ !enableStateStoreCheckpointIds
+
/**
* Whether the coordinator is reporting state stores trailing behind in
snapshot uploads.
*/
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManagerSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManagerSuite.scala
index 29d09f5d52f9..b15e8f167db5 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManagerSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ChecksumCheckpointFileManagerSuite.scala
@@ -26,6 +26,7 @@ import org.apache.hadoop.fs._
import org.apache.spark.SparkException
import org.apache.spark.sql.execution.streaming.checkpointing._
+import org.apache.spark.sql.internal.SQLConf
/**
* This inherits tests for the [[CheckpointFileManager]] from
[[CheckpointFileManagerTests]].
@@ -57,9 +58,27 @@ abstract class ChecksumCheckpointFileManagerSuite extends
CheckpointFileManagerT
s"expected main files: $mainFilesForExistingChecksumFiles / actual
files: $files")
}
+ override def createManager(path: Path): CheckpointFileManager = {
+ createChecksumManager(
+ path,
+ skipCreationIfFileMissingChecksum =
+
SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_SKIP_CREATION_IF_FILE_MISSING_CHECKSUM
+ .defaultValue.get)
+ }
+
/** Create a normal CheckpointFileManager (not the checksum checkpoint
manager) */
protected def createNoChecksumManager(path: Path): CheckpointFileManager
+ protected def createChecksumManager(
+ path: Path,
+ skipCreationIfFileMissingChecksum: Boolean): CheckpointFileManager = {
+ new ChecksumCheckpointFileManager(
+ createNoChecksumManager(path),
+ allowConcurrentDelete = true,
+ numThreads = 4,
+ skipCreationIfFileMissingChecksum = skipCreationIfFileMissingChecksum)
+ }
+
private def makeDir(fm: CheckpointFileManager, dir: Path): Unit = {
assert(!fm.exists(dir))
fm.mkdirs(dir)
@@ -184,29 +203,62 @@ abstract class ChecksumCheckpointFileManagerSuite extends
CheckpointFileManagerT
assert(regularFm.open(path).readContent() == content)
}
}
-}
-class FileContextChecksumCheckpointFileManagerSuite extends
ChecksumCheckpointFileManagerSuite {
- override def createManager(path: Path): CheckpointFileManager = {
- new ChecksumCheckpointFileManager(
- createNoChecksumManager(path),
- allowConcurrentDelete = true,
- numThreads = 4)
+ test("skip checksum creation if file missing checksum") {
+ withTempHadoopPath { basePath =>
+ val regularFm = createNoChecksumManager(basePath)
+ // Mkdirs
+ val dir = new Path(s"$basePath/dir/subdir/subsubdir")
+ makeDir(regularFm, dir)
+
+ // Create a file using the regular file manager
+ val path = new Path(s"$dir/file")
+ regularFm.createAtomic(path, overwriteIfPossible =
true).writeContent(content).close()
+ assert(regularFm.exists(path))
+
+ // Now try to write and read the file with the checksum manager with
fallback.
+ val checksumFmWithFallback =
+ createChecksumManager(basePath, skipCreationIfFileMissingChecksum =
true)
+ // Overwrite the file with a different content.
+ checksumFmWithFallback.createAtomic(
+ path, overwriteIfPossible = true).writeContent(content + 1).close()
+ assert(checksumFmWithFallback.open(path).readContent() == content + 1)
+ // Checksum should not be created since we fallback to the underlying
file manager.
+ assert(!checksumFmWithFallback.exists(getChecksumPath(path)))
+
+ // Now try to write and read the file with the checksum manager without
fallback.
+ val checksumFmWithoutFallback =
+ createChecksumManager(basePath, skipCreationIfFileMissingChecksum =
false)
+ // Overwrite the file with a different content.
+ checksumFmWithoutFallback.createAtomic(
+ path, overwriteIfPossible = true).writeContent(content + 2).close()
+ assert(checksumFmWithoutFallback.open(path).readContent() == content + 2)
+ // Checksum should be created since we don't fallback to the underlying
file manager.
+ assert(checksumFmWithoutFallback.exists(getChecksumPath(path)))
+
+ // Try to write and read the file with the checksum manager with
fallback when the checksum
+ // file already exists.
+ checksumFmWithFallback.createAtomic(
+ path, overwriteIfPossible = true).writeContent(content + 3).close()
+ // This read should succeed since we do not fallback to the underlying
file manager, since
+ // the checksum file already exists.
+ assert(checksumFmWithFallback.open(path).readContent() == content + 3)
+ assert(checksumFmWithFallback.exists(getChecksumPath(path)))
+
+ regularFm.close()
+ checksumFmWithFallback.close()
+ checksumFmWithoutFallback.close()
+ }
}
+}
+class FileContextChecksumCheckpointFileManagerSuite extends
ChecksumCheckpointFileManagerSuite {
protected def createNoChecksumManager(path: Path): CheckpointFileManager = {
new FileContextBasedCheckpointFileManager(path, new Configuration())
}
}
class FileSystemChecksumCheckpointFileManagerSuite extends
ChecksumCheckpointFileManagerSuite {
- override def createManager(path: Path): CheckpointFileManager = {
- new ChecksumCheckpointFileManager(
- createNoChecksumManager(path),
- allowConcurrentDelete = true,
- numThreads = 4)
- }
-
protected def createNoChecksumManager(path: Path): CheckpointFileManager = {
new FileSystemBasedCheckpointFileManager(path, new Configuration())
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala
index fad207f97dd9..898e324a954e 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/FailureInjectionCheckpointFileManager.scala
@@ -307,7 +307,8 @@ object FailureInjectionRocksDBStateStoreProvider {
localTempDir: File,
hadoopConf: Configuration,
codecName: String,
- loggingId: String): RocksDBFileManager = {
+ loggingId: String,
+ storeConf: StateStoreConf): RocksDBFileManager = {
new RocksDBFileManager(
dfsRootDir,
localTempDir,
@@ -315,7 +316,8 @@ object FailureInjectionRocksDBStateStoreProvider {
codecName,
loggingId = loggingId,
fileChecksumEnabled = this.conf.fileChecksumEnabled,
- fileChecksumThreadPoolSize = this.fileChecksumThreadPoolSize) {
+ fileChecksumThreadPoolSize = this.fileChecksumThreadPoolSize,
+ storeConf = this.conf.stateStoreConf) {
override def getFileSystem(
myDfsRootDir: String,
myHadoopConf: Configuration): FileSystem = {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala
index 30f1d77441bf..0b9690ee7277 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala
@@ -25,7 +25,7 @@ import org.apache.hadoop.conf.Configuration
import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
-import org.apache.spark.sql.functions.count
+import org.apache.spark.sql.functions.{count, udf}
import org.apache.spark.sql.internal.SQLConf
import
org.apache.spark.sql.internal.SQLConf.STREAMING_CHECKPOINT_FILE_MANAGER_CLASS
import org.apache.spark.sql.streaming._
@@ -587,6 +587,243 @@ class RocksDBCheckpointFailureInjectionSuite extends
StreamTest
}
}
+ case class FailureConf3(
+ skipCreationIfFileMissingChecksum: Boolean,
+ checkpointFormatVersion : String) {
+ override def toString: String = {
+ s"skipCreationIfFileMissingChecksum =
$skipCreationIfFileMissingChecksum, " +
+ s"checkpointFormatVersion = $checkpointFormatVersion"
+ }
+ }
+
+ private def versionsPresent(dir: File, suffix: String): Seq[(Long,
Option[String])] = {
+ dir.listFiles.filter(_.getName.endsWith(suffix))
+ .filter(!_.getName.startsWith("."))
+ .map(_.getName.stripSuffix(suffix).split("_"))
+ .map {
+ case Array(version, uniqueId) => (version.toLong, Some(uniqueId))
+ case Array(version) => (version.toLong, None)
+ }
+ .sorted
+ .distinct
+ .toSeq
+ }
+
+ /**
+ * Test that verifies upgrading from checksum disabled to checksum enabled
after state files are
+ * written but before batch commit completes. The important part of this
test is that files are
+ * not overwritten if they already exist. When checkpointFormatVersion is 2,
we will not run into
+ * the checksum verification failure because each batch run uses unique
changelog file names.
+ *
+ * Scenario:
+ * 1. Start with checksum verification disabled
+ * 2. Run batch 1 successfully (writes 1.changelog without .crc)
+ * 3. Start batch 2 - state store commits successfully (writes 2.changelog
without .crc) but the
+ * batch fails before the commit is complete (via UDF exception). This
leaves 2.changelog on
+ * disk without a corresponding commit log file
+ * 4. Restart query with checksum verification enabled and with a query
where the changelog file
+ * contents will change from batch 2
+ * 5. Run batch 2 again and it succeeds and writes 2.changelog (and
2.changelog.crc if
+ *
STREAMING_CHECKPOINT_FILE_CHECKSUM_SKIP_CREATION_IF_FILE_MISSING_CHECKSUM is
disabled)
+ * 6. Run batch 3 and it succeeds and writes 3.changelog and 3.changelog.crc
+ * 7. Query starts with
STREAMING_CHECKPOINT_FILE_CHECKSUM_SKIP_CREATION_IF_FILE_MISSING_CHECKSUM
+ * enabled/disabled and the different behavior is shown in the test
+ */
+
+ Seq(
+ FailureConf3(skipCreationIfFileMissingChecksum = false,
checkpointFormatVersion = "1"),
+ FailureConf3(skipCreationIfFileMissingChecksum = true,
checkpointFormatVersion = "1"),
+ FailureConf3(skipCreationIfFileMissingChecksum = false,
checkpointFormatVersion = "2"),
+ FailureConf3(skipCreationIfFileMissingChecksum = true,
checkpointFormatVersion = "2")
+ ).foreach { failureConf =>
+ test(s"Upgrading from file checksum disabled to enabled " +
+ "after state commits without batch commit " + failureConf.toString) {
+ val hadoopConf = new Configuration()
+ hadoopConf.set(STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key,
fileManagerClassName)
+ val rocksdbChangelogCheckpointingConfKey =
+ RocksDBConf.ROCKSDB_SQL_CONF_NAME_PREFIX +
".changelogCheckpointing.enabled"
+
+ withTempDirAllowFailureInjection { (checkpointDir, injectionState) =>
+ var forceTaskFailure = false
+ val failUDF = udf((value: Int) => {
+ if (forceTaskFailure) {
+ // This will fail all close() call to trigger query failures in
execution phase.
+ throw new RuntimeException("Ingest task failure")
+ }
+ value
+ })
+
+ val inputData = MemoryStream[Int]
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*").as("count"))
+ // would fail here after writing the changelog file for the agg
+ .select(failUDF($"value").as("value"), $"count")
+ .as[(Int, Long)]
+
+ val aggregated2 =
+ inputData.toDF()
+ .select($"value" + 1000 as "value") // This is to make the
changelog file different
+ .groupBy($"value")
+ .agg(count("*").as("count"))
+ // would fail here after writing the changelog file for the agg
+ .select(failUDF($"value").as("value"), $"count")
+ .as[(Int, Long)]
+
+ def getRunConf(checksumEnabled: Boolean) : Map[String, String] = {
+ Map(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
+ failureConf.checkpointFormatVersion,
+ SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED.key ->
checksumEnabled.toString,
+
SQLConf.STREAMING_CHECKPOINT_FILE_CHECKSUM_SKIP_CREATION_IF_FILE_MISSING_CHECKSUM.key
->
+ failureConf.skipCreationIfFileMissingChecksum.toString,
+ STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key ->
fileManagerClassName,
+ SQLConf.SHUFFLE_PARTITIONS.key -> "1")
+ }
+
+ // Verify that the changelog files exists for a version
+ def verifyChangelogFileExists(version: Long) : Boolean = {
+ versionsPresent(new File(checkpointDir, "state/0/0"),
".changelog").exists {
+ case (v, uniqueId) =>
+ if (failureConf.checkpointFormatVersion == "1") {
+ v == version && uniqueId.isEmpty
+ } else {
+ v == version && uniqueId.isDefined
+ }
+ }
+ }
+
+ // Verify that the changelog checksum files exists for a version
+ def verifyChangelogFileChecksumExists(version: Long) : Boolean = {
+ versionsPresent(new File(checkpointDir, "state/0/0"),
".changelog.crc").exists {
+ case (v, uniqueId) =>
+ if (failureConf.checkpointFormatVersion == "1") {
+ v == version && uniqueId.isEmpty
+ } else {
+ v == version && uniqueId.isDefined
+ }
+ }
+ }
+
+ // First run: file checksum disabled
+ val firstRunConfs = getRunConf(checksumEnabled = false)
+
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = firstRunConfs),
+ AddData(inputData, 3),
+ CheckLastBatch((3, 1)),
+ Execute { _ =>
+ forceTaskFailure = true
+ },
+ AddData(inputData, 3, 2),
+ ExpectFailure[SparkException] { ex =>
+ ex.getCause.getMessage.contains("FAILED_EXECUTE_UDF")
+ }
+ )
+
+ // Verify that the changelog file was written
+ assert(verifyChangelogFileExists(2))
+ // Verify that the changelog file checksum was NOT written since it
was disabled
+ assert(!verifyChangelogFileChecksumExists(2))
+
+ // Verify that the commit file was written
+ assert((new File(checkpointDir, "commits/0")).exists())
+ // Verify that the commit file was NOT written
+ assert(!(new File(checkpointDir, "commits/1")).exists())
+
+ // Second run: STREAMING_CHECKPOINT_FILE_CHECKSUM_ENABLED enabled with
+ // allowOverwriteInRename = false. This simulates an upgrade to a new
version where the
+ // file checksum is enabled. The allowOverwriteInRename is set to
false to test the case
+ // when overwriting the changelog file fails. This is to simulate the
case where the
+ // changelog file is not overwritten but the checksum file is written.
+ injectionState.allowOverwriteInRename = false
+ forceTaskFailure = false
+
+ val secondRunConfs = getRunConf(checksumEnabled = true)
+
+ inputData.addData(3, 1)
+
+ // The query should restart successfully and handle files without
checksums, whether
+ // skipCreationIfFileMissingChecksum is enabled or disabled. The
problem
+ // arises on the load after this run.
+ testStream(aggregated2, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = secondRunConfs),
+ AddData(inputData, 4),
+ CheckLastBatch((1003, 2), (1001, 1), (1004, 1)),
+ StopStream
+ )
+
+ assert(verifyChangelogFileExists(3))
+ assert(verifyChangelogFileChecksumExists(3))
+
+ // Verify that the commit files were written
+ assert((new File(checkpointDir, "commits/1")).exists())
+ assert((new File(checkpointDir, "commits/2")).exists())
+
+ val failureCase =
+ !failureConf.skipCreationIfFileMissingChecksum &&
+ failureConf.checkpointFormatVersion == "1"
+
+ if (failureCase) {
+ assert(verifyChangelogFileChecksumExists(2))
+
+ // The query does not succeed, since we load the old changelog file
with the checksum from
+ // the new changelog file that did not overwrite the old one. This
will lead to a checksum
+ // verification failure when we try to load the old changelog file
with the checksum from
+ // the new changelog file that did not overwrite the old one.
+ testStream(aggregated2, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = secondRunConfs),
+ AddData(inputData, 4),
+ ExpectFailure[SparkException] { ex =>
+
ex.getMessage.contains("CHECKPOINT_FILE_CHECKSUM_VERIFICATION_FAILED")
+ ex.getMessage.contains("2.changelog")
+ }
+ )
+
+ // Verify that the commit file was not written
+ assert(!(new File(checkpointDir, "commits/3")).exists())
+ } else {
+ if (failureConf.checkpointFormatVersion == "1") {
+ // With checkpointFormatVersion = 1, the changelog file checksum
should not be written
+ assert(!verifyChangelogFileChecksumExists(2))
+ } else {
+ // With checkpointFormatVersion = 2, the changelog file checksum
should be written
+ assert(verifyChangelogFileChecksumExists(2))
+ }
+
+ // The query should restart successfully
+ testStream(aggregated2, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = secondRunConfs),
+ AddData(inputData, 4),
+ CheckLastBatch((1004, 2)),
+ StopStream
+ )
+
+ // Verify again the 2.changelog file checksum exists or not
+ if (failureConf.checkpointFormatVersion == "1") {
+ assert(!verifyChangelogFileChecksumExists(2))
+ } else {
+ assert(verifyChangelogFileChecksumExists(2))
+ }
+
+ assert(verifyChangelogFileExists(4))
+ assert(verifyChangelogFileChecksumExists(4))
+ assert((new File(checkpointDir, "commits/3")).exists())
+ }
+ }
+ }
+ }
+
def commitAndGetCheckpointId(db: RocksDB): Option[String] = {
val (v, ci) = db.commit()
ci.stateStoreCkptId
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
