You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "siying (via GitHub)" <gi...@apache.org> on 2025/04/10 22:27:01 UTC
[PR] [SPARK-51768][SS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
siying opened a new pull request, #50559:
URL: https://github.com/apache/spark/pull/50559
### What changes were proposed in this pull request?
Add unit test to verify stream query works as expected when writing to commit or offset log fails
### Why are the changes needed?
Improve testing
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Run this existing test
### Was this patch authored or co-authored using generative AI tooling?
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2043006686
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -414,6 +414,84 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
}
}
+ case class FailureConf2(logType: String, checkpointFormatVersion: String) {
+ override def toString: String = {
+ s"logType = $logType, checkpointFormatVersion = $checkpointFormatVersion"
+ }
+ }
+
+ // tests to validate the behavior after failures when writing to the commit and offset logs
+ Seq(
+ FailureConf2("commits", checkpointFormatVersion = "1"),
+ FailureConf2("commits", checkpointFormatVersion = "2"),
+ FailureConf2("offsets", checkpointFormatVersion = "1"),
+ FailureConf2("offsets", checkpointFormatVersion = "2")).foreach { failureConf =>
+ test(s"Progress log fails to write $failureConf") {
+ 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) =>
+ withSQLConf(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> "2") {
+ val inputData = MemoryStream[Int]
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*"))
+ .as[(Int, Long)]
+
+ // This should cause the second batch to fail
+ injectionState.createAtomicDelayCloseRegex = Seq(s".*/${failureConf.logType}/1")
+
+ val additionalConfs = Map(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
+ failureConf.checkpointFormatVersion,
+ STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fileManagerClassName)
+
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 3),
+ CheckLastBatch((3, 1)),
+ AddData(inputData, 3, 2),
+ // We should categorize this error.
+ // TODO after the error is categorized, we should check error class
+ ExpectFailure[IOException] { _ => () }
+ )
+
+ injectionState.createAtomicDelayCloseRegex = Seq.empty
+
+ inputData.addData(3, 1)
+
+ // The query will restart successfully and start at the checkpoint after Batch 1
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 4),
+ if (failureConf.logType == "commits") {
+ // If the failure is in the commit log, data is already committed. The batch will
Review Comment:
This is very uneasy to follow, because the behavior of MemoryStream seems to impact a lot on the test. It's not easy to reason about when the commit on the source happens and how MemoryStream will behave. It'd be harder than the origin test logic.
I'd test with file stream where we only append files and Spark is expected to process all files "regardless of these failures". (This should be a contract.) Once we change the output mode to complete, we should see the same result in the latest batch which has no further file to process.
Also, I'm not comfortable with the behavior - `If the failure is in the commit log, data is already committed.`. Shouldn't we commit the offset for batch N to source when batch N is NOT committed? I suspect this is an indication of data loss/correctness - I wish I'm misunderstanding something.
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2043090343
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -414,6 +414,84 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
}
}
+ case class FailureConf2(logType: String, checkpointFormatVersion: String) {
+ override def toString: String = {
+ s"logType = $logType, checkpointFormatVersion = $checkpointFormatVersion"
+ }
+ }
+
+ // tests to validate the behavior after failures when writing to the commit and offset logs
+ Seq(
+ FailureConf2("commits", checkpointFormatVersion = "1"),
+ FailureConf2("commits", checkpointFormatVersion = "2"),
+ FailureConf2("offsets", checkpointFormatVersion = "1"),
+ FailureConf2("offsets", checkpointFormatVersion = "2")).foreach { failureConf =>
+ test(s"Progress log fails to write $failureConf") {
+ 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) =>
+ withSQLConf(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> "2") {
+ val inputData = MemoryStream[Int]
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*"))
+ .as[(Int, Long)]
+
+ // This should cause the second batch to fail
+ injectionState.createAtomicDelayCloseRegex = Seq(s".*/${failureConf.logType}/1")
+
+ val additionalConfs = Map(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
+ failureConf.checkpointFormatVersion,
+ STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fileManagerClassName)
+
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 3),
+ CheckLastBatch((3, 1)),
+ AddData(inputData, 3, 2),
+ // We should categorize this error.
+ // TODO after the error is categorized, we should check error class
+ ExpectFailure[IOException] { _ => () }
+ )
+
+ injectionState.createAtomicDelayCloseRegex = Seq.empty
+
+ inputData.addData(3, 1)
+
+ // The query will restart successfully and start at the checkpoint after Batch 1
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 4),
+ if (failureConf.logType == "commits") {
+ // If the failure is in the commit log, data is already committed. The batch will
Review Comment:
If that's hard to achieve, I'm OK with more direct code comment about "sink" status - e.g. (3, 2) and (2, 1) were already emitted to sink as batch 1 even with failure on commit log. The write against sink for retried batch 1 should have been ignored (memory sink), but we restart the query and the state in the sink has reset, hence the write against sink for batch 1 in new query has effect.
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #50559:
URL: https://github.com/apache/spark/pull/50559#issuecomment-2807755306
Thanks! Merging to master.
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2045581330
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -475,14 +475,14 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
additionalConfs = additionalConfs),
AddData(inputData, 4),
if (failureConf.logType == "commits") {
- // If the failure is in the commit log, data is already committed. The batch will
- // rerun but the comit will be no-op.
- CheckLastBatch((3, 3), (1, 1), (4, 1))
+ // If the failure is in the commit log, data is already committed.
+ // MemorySink isn't a EactlyOnce sink, so we will see the data from the previous
Review Comment:
```suggestion
// MemorySink isn't an ExactlyOnce sink, so we will see the data from the previous
```
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2045582101
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -475,14 +475,14 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
additionalConfs = additionalConfs),
AddData(inputData, 4),
if (failureConf.logType == "commits") {
- // If the failure is in the commit log, data is already committed. The batch will
- // rerun but the comit will be no-op.
- CheckLastBatch((3, 3), (1, 1), (4, 1))
+ // If the failure is in the commit log, data is already committed.
+ // MemorySink isn't a EactlyOnce sink, so we will see the data from the previous
Review Comment:
Btw it is exactly-once "under single run".
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2043090343
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -414,6 +414,84 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
}
}
+ case class FailureConf2(logType: String, checkpointFormatVersion: String) {
+ override def toString: String = {
+ s"logType = $logType, checkpointFormatVersion = $checkpointFormatVersion"
+ }
+ }
+
+ // tests to validate the behavior after failures when writing to the commit and offset logs
+ Seq(
+ FailureConf2("commits", checkpointFormatVersion = "1"),
+ FailureConf2("commits", checkpointFormatVersion = "2"),
+ FailureConf2("offsets", checkpointFormatVersion = "1"),
+ FailureConf2("offsets", checkpointFormatVersion = "2")).foreach { failureConf =>
+ test(s"Progress log fails to write $failureConf") {
+ 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) =>
+ withSQLConf(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> "2") {
+ val inputData = MemoryStream[Int]
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*"))
+ .as[(Int, Long)]
+
+ // This should cause the second batch to fail
+ injectionState.createAtomicDelayCloseRegex = Seq(s".*/${failureConf.logType}/1")
+
+ val additionalConfs = Map(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
+ failureConf.checkpointFormatVersion,
+ STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fileManagerClassName)
+
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 3),
+ CheckLastBatch((3, 1)),
+ AddData(inputData, 3, 2),
+ // We should categorize this error.
+ // TODO after the error is categorized, we should check error class
+ ExpectFailure[IOException] { _ => () }
+ )
+
+ injectionState.createAtomicDelayCloseRegex = Seq.empty
+
+ inputData.addData(3, 1)
+
+ // The query will restart successfully and start at the checkpoint after Batch 1
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 4),
+ if (failureConf.logType == "commits") {
+ // If the failure is in the commit log, data is already committed. The batch will
Review Comment:
If that's hard to achieve, I'm OK with more direct code comment about "sink" status - e.g. (3, 2) and (2, 1) were already emitted to sink as batch 1 even with failure on commit log. The write against sink for batch 1 should have been ignored (memory sink), but we restart the query and the state in the sink has reset, hence the write against sink for batch 1 in new query has effect.
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2043086506
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -414,6 +414,84 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
}
}
+ case class FailureConf2(logType: String, checkpointFormatVersion: String) {
+ override def toString: String = {
+ s"logType = $logType, checkpointFormatVersion = $checkpointFormatVersion"
+ }
+ }
+
+ // tests to validate the behavior after failures when writing to the commit and offset logs
+ Seq(
+ FailureConf2("commits", checkpointFormatVersion = "1"),
+ FailureConf2("commits", checkpointFormatVersion = "2"),
+ FailureConf2("offsets", checkpointFormatVersion = "1"),
+ FailureConf2("offsets", checkpointFormatVersion = "2")).foreach { failureConf =>
+ test(s"Progress log fails to write $failureConf") {
+ 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) =>
+ withSQLConf(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> "2") {
+ val inputData = MemoryStream[Int]
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*"))
+ .as[(Int, Long)]
+
+ // This should cause the second batch to fail
+ injectionState.createAtomicDelayCloseRegex = Seq(s".*/${failureConf.logType}/1")
+
+ val additionalConfs = Map(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
+ failureConf.checkpointFormatVersion,
+ STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fileManagerClassName)
+
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 3),
+ CheckLastBatch((3, 1)),
+ AddData(inputData, 3, 2),
+ // We should categorize this error.
+ // TODO after the error is categorized, we should check error class
+ ExpectFailure[IOException] { _ => () }
+ )
+
+ injectionState.createAtomicDelayCloseRegex = Seq.empty
+
+ inputData.addData(3, 1)
+
+ // The query will restart successfully and start at the checkpoint after Batch 1
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 4),
+ if (failureConf.logType == "commits") {
+ // If the failure is in the commit log, data is already committed. The batch will
Review Comment:
Wait, is it due to the fact we used Check"Last"Batch? Sigh I missed this.
I'm OK with not using file stream, but let's figure out how we can verify that (2, 1) has provided before this batch.
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #50559: [SPARK-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures
URL: https://github.com/apache/spark/pull/50559
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "siying (via GitHub)" <gi...@apache.org>.
siying commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2045127704
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -414,6 +414,84 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
}
}
+ case class FailureConf2(logType: String, checkpointFormatVersion: String) {
+ override def toString: String = {
+ s"logType = $logType, checkpointFormatVersion = $checkpointFormatVersion"
+ }
+ }
+
+ // tests to validate the behavior after failures when writing to the commit and offset logs
+ Seq(
+ FailureConf2("commits", checkpointFormatVersion = "1"),
+ FailureConf2("commits", checkpointFormatVersion = "2"),
+ FailureConf2("offsets", checkpointFormatVersion = "1"),
+ FailureConf2("offsets", checkpointFormatVersion = "2")).foreach { failureConf =>
+ test(s"Progress log fails to write $failureConf") {
+ 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) =>
+ withSQLConf(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> "2") {
+ val inputData = MemoryStream[Int]
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*"))
+ .as[(Int, Long)]
+
+ // This should cause the second batch to fail
+ injectionState.createAtomicDelayCloseRegex = Seq(s".*/${failureConf.logType}/1")
+
+ val additionalConfs = Map(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
+ failureConf.checkpointFormatVersion,
+ STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fileManagerClassName)
+
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 3),
+ CheckLastBatch((3, 1)),
+ AddData(inputData, 3, 2),
+ // We should categorize this error.
+ // TODO after the error is categorized, we should check error class
+ ExpectFailure[IOException] { _ => () }
+ )
+
+ injectionState.createAtomicDelayCloseRegex = Seq.empty
+
+ inputData.addData(3, 1)
+
+ // The query will restart successfully and start at the checkpoint after Batch 1
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 4),
+ if (failureConf.logType == "commits") {
+ // If the failure is in the commit log, data is already committed. The batch will
Review Comment:
Sorry I think I by mistake used CheckLastBatch. Using CheckNewAnswer will be much less confusing and I'm changing to use 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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "siying (via GitHub)" <gi...@apache.org>.
siying commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2043073111
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -414,6 +414,84 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
}
}
+ case class FailureConf2(logType: String, checkpointFormatVersion: String) {
+ override def toString: String = {
+ s"logType = $logType, checkpointFormatVersion = $checkpointFormatVersion"
+ }
+ }
+
+ // tests to validate the behavior after failures when writing to the commit and offset logs
+ Seq(
+ FailureConf2("commits", checkpointFormatVersion = "1"),
+ FailureConf2("commits", checkpointFormatVersion = "2"),
+ FailureConf2("offsets", checkpointFormatVersion = "1"),
+ FailureConf2("offsets", checkpointFormatVersion = "2")).foreach { failureConf =>
+ test(s"Progress log fails to write $failureConf") {
+ 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) =>
+ withSQLConf(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> "2") {
+ val inputData = MemoryStream[Int]
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*"))
+ .as[(Int, Long)]
+
+ // This should cause the second batch to fail
+ injectionState.createAtomicDelayCloseRegex = Seq(s".*/${failureConf.logType}/1")
+
+ val additionalConfs = Map(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
+ failureConf.checkpointFormatVersion,
+ STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fileManagerClassName)
+
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 3),
+ CheckLastBatch((3, 1)),
+ AddData(inputData, 3, 2),
+ // We should categorize this error.
+ // TODO after the error is categorized, we should check error class
+ ExpectFailure[IOException] { _ => () }
+ )
+
+ injectionState.createAtomicDelayCloseRegex = Seq.empty
+
+ inputData.addData(3, 1)
+
+ // The query will restart successfully and start at the checkpoint after Batch 1
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 4),
+ if (failureConf.logType == "commits") {
+ // If the failure is in the commit log, data is already committed. The batch will
Review Comment:
I'll try to clean up the test code to be clearer, but committing source shouldn't be related here. We only cleanup the offset for the previous batch before ending the current one, so it should have no effect.
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #50559:
URL: https://github.com/apache/spark/pull/50559#issuecomment-2807755010
https://github.com/siying/spark/actions/runs/14479972406/job/40614700665
pyspark-connect is only failing (with OOME), which is not relevant to this change.
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2043100080
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -414,6 +414,84 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
}
}
+ case class FailureConf2(logType: String, checkpointFormatVersion: String) {
+ override def toString: String = {
+ s"logType = $logType, checkpointFormatVersion = $checkpointFormatVersion"
+ }
+ }
+
+ // tests to validate the behavior after failures when writing to the commit and offset logs
+ Seq(
+ FailureConf2("commits", checkpointFormatVersion = "1"),
+ FailureConf2("commits", checkpointFormatVersion = "2"),
+ FailureConf2("offsets", checkpointFormatVersion = "1"),
+ FailureConf2("offsets", checkpointFormatVersion = "2")).foreach { failureConf =>
+ test(s"Progress log fails to write $failureConf") {
+ 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) =>
+ withSQLConf(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> "2") {
+ val inputData = MemoryStream[Int]
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*"))
+ .as[(Int, Long)]
+
+ // This should cause the second batch to fail
+ injectionState.createAtomicDelayCloseRegex = Seq(s".*/${failureConf.logType}/1")
+
+ val additionalConfs = Map(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
+ failureConf.checkpointFormatVersion,
+ STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fileManagerClassName)
+
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 3),
+ CheckLastBatch((3, 1)),
+ AddData(inputData, 3, 2),
+ // We should categorize this error.
+ // TODO after the error is categorized, we should check error class
+ ExpectFailure[IOException] { _ => () }
+ )
+
+ injectionState.createAtomicDelayCloseRegex = Seq.empty
+
+ inputData.addData(3, 1)
+
+ // The query will restart successfully and start at the checkpoint after Batch 1
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 4),
+ if (failureConf.logType == "commits") {
+ // If the failure is in the commit log, data is already committed. The batch will
Review Comment:
It'd be lot easier if we can describe the data being included in batch 0 to N. This is very complicated because I expect (3, 2) to be "reprocessed" in the "next query run", and we don't see it anywhere despite the fact the sink has reset hence there shouldn't be dedup for batch 1.
It seems to be due to CheckLastBatch - if we use CheckAnswer then it should contain both. If there is no easy way to verify two batches separately, let's just use CheckAnswer to confirm both batches altogether, with a code comment to describe we reprocessed (3, 2) in batch 1, and processed further input in batch 2.
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2043006686
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -414,6 +414,84 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
}
}
+ case class FailureConf2(logType: String, checkpointFormatVersion: String) {
+ override def toString: String = {
+ s"logType = $logType, checkpointFormatVersion = $checkpointFormatVersion"
+ }
+ }
+
+ // tests to validate the behavior after failures when writing to the commit and offset logs
+ Seq(
+ FailureConf2("commits", checkpointFormatVersion = "1"),
+ FailureConf2("commits", checkpointFormatVersion = "2"),
+ FailureConf2("offsets", checkpointFormatVersion = "1"),
+ FailureConf2("offsets", checkpointFormatVersion = "2")).foreach { failureConf =>
+ test(s"Progress log fails to write $failureConf") {
+ 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) =>
+ withSQLConf(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> "2") {
+ val inputData = MemoryStream[Int]
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*"))
+ .as[(Int, Long)]
+
+ // This should cause the second batch to fail
+ injectionState.createAtomicDelayCloseRegex = Seq(s".*/${failureConf.logType}/1")
+
+ val additionalConfs = Map(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
+ failureConf.checkpointFormatVersion,
+ STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fileManagerClassName)
+
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 3),
+ CheckLastBatch((3, 1)),
+ AddData(inputData, 3, 2),
+ // We should categorize this error.
+ // TODO after the error is categorized, we should check error class
+ ExpectFailure[IOException] { _ => () }
+ )
+
+ injectionState.createAtomicDelayCloseRegex = Seq.empty
+
+ inputData.addData(3, 1)
+
+ // The query will restart successfully and start at the checkpoint after Batch 1
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 4),
+ if (failureConf.logType == "commits") {
+ // If the failure is in the commit log, data is already committed. The batch will
Review Comment:
This is very uneasy to follow, because the behavior of MemoryStream seems to impact a lot on the test. It's not easy to reason about when the commit on the source happens and how MemoryStream will behave. It'd be harder than the origin test logic.
I'd test with file stream where we only append files and Spark is expected to process all files "regardless of these failures". (This should be a contract.) Once we change the output mode to complete, we should see the same result in the latest batch which processes all files.
Also, I'm not comfortable with the behavior - `If the failure is in the commit log, data is already committed.`. Shouldn't we commit the offset for batch N to source when batch N is NOT committed? I suspect this is an indication of data loss/correctness - I wish I'm misunderstanding something.
--
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-51768][SS][TESTS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on code in PR #50559:
URL: https://github.com/apache/spark/pull/50559#discussion_r2043006686
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBCheckpointFailureInjectionSuite.scala:
##########
@@ -414,6 +414,84 @@ class RocksDBCheckpointFailureInjectionSuite extends StreamTest
}
}
+ case class FailureConf2(logType: String, checkpointFormatVersion: String) {
+ override def toString: String = {
+ s"logType = $logType, checkpointFormatVersion = $checkpointFormatVersion"
+ }
+ }
+
+ // tests to validate the behavior after failures when writing to the commit and offset logs
+ Seq(
+ FailureConf2("commits", checkpointFormatVersion = "1"),
+ FailureConf2("commits", checkpointFormatVersion = "2"),
+ FailureConf2("offsets", checkpointFormatVersion = "1"),
+ FailureConf2("offsets", checkpointFormatVersion = "2")).foreach { failureConf =>
+ test(s"Progress log fails to write $failureConf") {
+ 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) =>
+ withSQLConf(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_MIN_DELTAS_FOR_SNAPSHOT.key -> "2") {
+ val inputData = MemoryStream[Int]
+ val aggregated =
+ inputData.toDF()
+ .groupBy($"value")
+ .agg(count("*"))
+ .as[(Int, Long)]
+
+ // This should cause the second batch to fail
+ injectionState.createAtomicDelayCloseRegex = Seq(s".*/${failureConf.logType}/1")
+
+ val additionalConfs = Map(
+ rocksdbChangelogCheckpointingConfKey -> "true",
+ SQLConf.STATE_STORE_CHECKPOINT_FORMAT_VERSION.key ->
+ failureConf.checkpointFormatVersion,
+ STREAMING_CHECKPOINT_FILE_MANAGER_CLASS.parent.key -> fileManagerClassName)
+
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 3),
+ CheckLastBatch((3, 1)),
+ AddData(inputData, 3, 2),
+ // We should categorize this error.
+ // TODO after the error is categorized, we should check error class
+ ExpectFailure[IOException] { _ => () }
+ )
+
+ injectionState.createAtomicDelayCloseRegex = Seq.empty
+
+ inputData.addData(3, 1)
+
+ // The query will restart successfully and start at the checkpoint after Batch 1
+ testStream(aggregated, Update)(
+ StartStream(
+ checkpointLocation = checkpointDir.getAbsolutePath,
+ additionalConfs = additionalConfs),
+ AddData(inputData, 4),
+ if (failureConf.logType == "commits") {
+ // If the failure is in the commit log, data is already committed. The batch will
Review Comment:
This is very uneasy to follow, because the behavior of MemoryStream seems to impact a lot on the test. It's not easy to reason about when the commit on the source happens and how MemoryStream will behave. It'd be harder than the origin test logic.
I'd test with file stream where we only append files and Spark is expected to process all files "regardless of these failures". Once we change the output mode to complete, we should see the same result in the latest batch which processes all files.
Also, I'm not comfortable with the behavior - `If the failure is in the commit log, data is already committed.`. Shouldn't we commit the offset for batch N to source when batch N is NOT committed? I suspect this is an indication of data loss/correctness - I wish I'm misunderstanding something.
--
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-51768][SS] Create Failure Injection Test for Streaming offset and commit log write failures [spark]
Posted by "siying (via GitHub)" <gi...@apache.org>.
siying commented on PR #50559:
URL: https://github.com/apache/spark/pull/50559#issuecomment-2795308453
@HeartSaVioR this PR is very simple. Can you take a look?
--
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