You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/02/14 11:23:08 UTC
[GitHub] [spark] LeeeeLiu opened a new pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
LeeeeLiu opened a new pull request #35513:
URL: https://github.com/apache/spark/pull/35513
…mitId and offsetId are inconsistent
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### 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.
-->
The code of method: populateStartOffsets in class: org.apache.spark.sql.execution.streaming.MicroBatchExecution is modified.
### 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.
-->
In some unexpected cases, commit and offset are inconsistent, and offset is not written into HDFS continuously, as follows:
commits
/tmp/streaming_xxxxxxxx/commits/113256
/tmp/streaming_xxxxxxxx/commits/113257
offsets
/tmp/streaming_xxxxxxxx/offsets/113257
/tmp/streaming_xxxxxxxx/offsets/113259
When we start the streaming program, batch ${latestBatchId - 1} is 113258, but offsets 113258 doesn't exist, an exception will be thrown, resulting in the program cannot be started. As an improvement, Spark doesn‘t need to repair itself, but we could probably do some simply analysis and give better error message.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
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'.
-->
Yes.
An error message is logged if the exception is thrown.
### 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.
-->
I have provided a test case that can output logs correctly.
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35513:
URL: https://github.com/apache/spark/pull/35513#discussion_r805746642
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecutionSuite.scala
##########
@@ -153,7 +179,6 @@ class MicroBatchExecutionSuite extends StreamTest with BeforeAndAfter {
)
}
-
Review comment:
nit: let's revert the unnecessary change
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecutionSuite.scala
##########
@@ -74,6 +78,27 @@ class MicroBatchExecutionSuite extends StreamTest with BeforeAndAfter {
)
}
+ test("SPARK-38033: SS cannot be started because the commitId and offsetId are inconsistent") {
+ val inputData = MemoryStream[Int]
+ val streamEvent = inputData.toDF().select("value")
+
+ val resourceUri = this.getClass.getResource(
+ "/structured-streaming/checkpoint-test-offsetId-commitId-inconsistent/").toURI
+
+ val checkpointDir = Utils.createTempDir().getCanonicalFile
+ // Copy the checkpoint to a temp dir to prevent changes to the original.
+ // Not doing this will lead to the test passing on the first run, but fail subsequent runs.
+ FileUtils.copyDirectory(new File(resourceUri), checkpointDir)
+
+ testStream(streamEvent) (
+ AddData(inputData, 1, 2, 3, 4, 5, 6),
+ StartStream(Trigger.Once, checkpointLocation = checkpointDir.getAbsolutePath),
+ ExpectFailure[IllegalStateException](e => {
Review comment:
nit: you can use `{ e =>` instead to remove ()
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35513:
URL: https://github.com/apache/spark/pull/35513#issuecomment-1054331146
+1
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
[GitHub] [spark] LeeeeLiu commented on a change in pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
LeeeeLiu commented on a change in pull request #35513:
URL: https://github.com/apache/spark/pull/35513#discussion_r807475288
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecutionSuite.scala
##########
@@ -153,7 +179,6 @@ class MicroBatchExecutionSuite extends StreamTest with BeforeAndAfter {
)
}
-
Review comment:
Okay, thank you. I'll fix it.
--
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
[GitHub] [spark] LeeeeLiu commented on pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
LeeeeLiu commented on pull request #35513:
URL: https://github.com/apache/spark/pull/35513#issuecomment-1041034503
> Assuming you've verified this fix manually, could you please produce the step to reproduce, and the exact log message you got? It would be nice if we have it in the section of "How was this patch tested?".
>
> Your new test doesn't test the new change (although it's hard to test against log message), so the manual verification seems to be required.
Hi @HeartSaVioR , I will give detailed test instructions in the section of "How was this patch tested?".
I actually tested the original exception in the new test. Because I don't know how to test for error messages. However, when an exception is thrown, a newly added error message is printed. So I think this new test tests a new change.
I have manually compiled the master branch to simulate, and the error messages will also output normally.
--
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
[GitHub] [spark] LeeeeLiu commented on a change in pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
LeeeeLiu commented on a change in pull request #35513:
URL: https://github.com/apache/spark/pull/35513#discussion_r815709049
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
##########
@@ -314,7 +314,27 @@ class MicroBatchExecution(
/* Initialize committed offsets to a committed batch, which at this
* is the second latest batch id in the offset log. */
if (latestBatchId != 0) {
+ /* SPARK-38033: In some unexpected cases, commit and offset are inconsistent,
+ * and offset is not written into HDFS continuously, as follows:
+ *
+ * commits
+ * /tmp/streaming_xxxxxxxx/commits/113256
+ * /tmp/streaming_xxxxxxxx/commits/113257
+ * offsets
+ * /tmp/streaming_xxxxxxxx/offsets/113257
+ * /tmp/streaming_xxxxxxxx/offsets/113259
+ *
+ * When we start the streaming program, batch ${latestBatchId - 1} is 113258,
+ * but offsets 113258 doesn't exist, an exception will be thrown,resulting in
+ * the program cannot be started. As an improvement, we could probably do some
+ * simply analysis and give better error message. */
val secondLatestOffsets = offsetLog.get(latestBatchId - 1).getOrElse {
+ logError(s"Please check the checkpoint, batch ${latestBatchId - 1} doesn't exist. " +
Review comment:
OK, I agree with you. Let me change it.
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35513:
URL: https://github.com/apache/spark/pull/35513#discussion_r815579852
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
##########
@@ -314,7 +314,27 @@ class MicroBatchExecution(
/* Initialize committed offsets to a committed batch, which at this
* is the second latest batch id in the offset log. */
if (latestBatchId != 0) {
+ /* SPARK-38033: In some unexpected cases, commit and offset are inconsistent,
+ * and offset is not written into HDFS continuously, as follows:
+ *
+ * commits
+ * /tmp/streaming_xxxxxxxx/commits/113256
+ * /tmp/streaming_xxxxxxxx/commits/113257
+ * offsets
+ * /tmp/streaming_xxxxxxxx/offsets/113257
+ * /tmp/streaming_xxxxxxxx/offsets/113259
+ *
+ * When we start the streaming program, batch ${latestBatchId - 1} is 113258,
+ * but offsets 113258 doesn't exist, an exception will be thrown,resulting in
+ * the program cannot be started. As an improvement, we could probably do some
+ * simply analysis and give better error message. */
val secondLatestOffsets = offsetLog.get(latestBatchId - 1).getOrElse {
+ logError(s"Please check the checkpoint, batch ${latestBatchId - 1} doesn't exist. " +
Review comment:
Since there is no feedback, shall we pick my suggestion for the initial version till someone brings up better sentence?
--
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
[GitHub] [spark] AmplabJenkins commented on pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #35513:
URL: https://github.com/apache/spark/pull/35513#issuecomment-1039683113
Can one of the admins verify this patch?
--
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
[GitHub] [spark] LeeeeLiu commented on pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
LeeeeLiu commented on pull request #35513:
URL: https://github.com/apache/spark/pull/35513#issuecomment-1039012524
Hi, @HyukjinKwon , @HeartSaVioR, this is a problem I've already verified in the master branch and provide a test case that can output logs correctly. Could you please verify this patch?
--
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
[GitHub] [spark] LeeeeLiu commented on pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
LeeeeLiu commented on pull request #35513:
URL: https://github.com/apache/spark/pull/35513#issuecomment-1039013877
Fixes based on Branch -3.1 points: [https://github.com/apache/spark/pull/35458](url)
--
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
[GitHub] [spark] HeartSaVioR commented on pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35513:
URL: https://github.com/apache/spark/pull/35513#issuecomment-1054333127
@LeeeeLiu Thanks for your contribution! I merged this into 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
[GitHub] [spark] HeartSaVioR closed pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
HeartSaVioR closed pull request #35513:
URL: https://github.com/apache/spark/pull/35513
--
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
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35513:
URL: https://github.com/apache/spark/pull/35513#discussion_r808404922
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
##########
@@ -314,7 +314,27 @@ class MicroBatchExecution(
/* Initialize committed offsets to a committed batch, which at this
* is the second latest batch id in the offset log. */
if (latestBatchId != 0) {
+ /* SPARK-38033: In some unexpected cases, commit and offset are inconsistent,
Review comment:
We wouldn't need this code comment. The log message we add should be self-explanatory.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
##########
@@ -314,7 +314,27 @@ class MicroBatchExecution(
/* Initialize committed offsets to a committed batch, which at this
* is the second latest batch id in the offset log. */
if (latestBatchId != 0) {
+ /* SPARK-38033: In some unexpected cases, commit and offset are inconsistent,
+ * and offset is not written into HDFS continuously, as follows:
+ *
+ * commits
+ * /tmp/streaming_xxxxxxxx/commits/113256
+ * /tmp/streaming_xxxxxxxx/commits/113257
+ * offsets
+ * /tmp/streaming_xxxxxxxx/offsets/113257
+ * /tmp/streaming_xxxxxxxx/offsets/113259
+ *
+ * When we start the streaming program, batch ${latestBatchId - 1} is 113258,
+ * but offsets 113258 doesn't exist, an exception will be thrown,resulting in
+ * the program cannot be started. As an improvement, we could probably do some
+ * simply analysis and give better error message. */
val secondLatestOffsets = offsetLog.get(latestBatchId - 1).getOrElse {
+ logError(s"Please check the checkpoint, batch ${latestBatchId - 1} doesn't exist. " +
Review comment:
I guess we can adjust the log message like below to give more context and direct to what end users should do:
> s"The offset log for batch ${latestBatchId - 1} doesn't exist, which is required to restart the query from the latest batch $latestBatchId from the offset log. Please ensure there are two subsequent offset logs available for the latest batch via manually deleting the offset file(s). Please also ensure the latest batch for commit log is equal or one batch earlier than the latest batch for offset log."
cc. @viirya @xuanyuanking Appreciate your refinement or better idea on guidance message. Thanks!
--
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
[GitHub] [spark] LeeeeLiu commented on a change in pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
LeeeeLiu commented on a change in pull request #35513:
URL: https://github.com/apache/spark/pull/35513#discussion_r813721750
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
##########
@@ -314,7 +314,27 @@ class MicroBatchExecution(
/* Initialize committed offsets to a committed batch, which at this
* is the second latest batch id in the offset log. */
if (latestBatchId != 0) {
+ /* SPARK-38033: In some unexpected cases, commit and offset are inconsistent,
+ * and offset is not written into HDFS continuously, as follows:
+ *
+ * commits
+ * /tmp/streaming_xxxxxxxx/commits/113256
+ * /tmp/streaming_xxxxxxxx/commits/113257
+ * offsets
+ * /tmp/streaming_xxxxxxxx/offsets/113257
+ * /tmp/streaming_xxxxxxxx/offsets/113259
+ *
+ * When we start the streaming program, batch ${latestBatchId - 1} is 113258,
+ * but offsets 113258 doesn't exist, an exception will be thrown,resulting in
+ * the program cannot be started. As an improvement, we could probably do some
+ * simply analysis and give better error message. */
val secondLatestOffsets = offsetLog.get(latestBatchId - 1).getOrElse {
+ logError(s"Please check the checkpoint, batch ${latestBatchId - 1} doesn't exist. " +
Review comment:
cc. @HeartSaVioR @cloud-fan @dongjoon-hyun . Could you help provide some guidance advice? Thanks so much!
--
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
[GitHub] [spark] LeeeeLiu commented on a change in pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
LeeeeLiu commented on a change in pull request #35513:
URL: https://github.com/apache/spark/pull/35513#discussion_r808625180
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
##########
@@ -314,7 +314,27 @@ class MicroBatchExecution(
/* Initialize committed offsets to a committed batch, which at this
* is the second latest batch id in the offset log. */
if (latestBatchId != 0) {
+ /* SPARK-38033: In some unexpected cases, commit and offset are inconsistent,
Review comment:
Okay, I will delete this code comment in the new commit. Thanks.
--
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
[GitHub] [spark] AmplabJenkins commented on pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #35513:
URL: https://github.com/apache/spark/pull/35513#issuecomment-1039683113
Can one of the admins verify this patch?
--
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
[GitHub] [spark] LeeeeLiu commented on a change in pull request #35513: [SPARK-38033][SS] The SS processing cannot be started because the com…
Posted by GitBox <gi...@apache.org>.
LeeeeLiu commented on a change in pull request #35513:
URL: https://github.com/apache/spark/pull/35513#discussion_r807475460
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecutionSuite.scala
##########
@@ -74,6 +78,27 @@ class MicroBatchExecutionSuite extends StreamTest with BeforeAndAfter {
)
}
+ test("SPARK-38033: SS cannot be started because the commitId and offsetId are inconsistent") {
+ val inputData = MemoryStream[Int]
+ val streamEvent = inputData.toDF().select("value")
+
+ val resourceUri = this.getClass.getResource(
+ "/structured-streaming/checkpoint-test-offsetId-commitId-inconsistent/").toURI
+
+ val checkpointDir = Utils.createTempDir().getCanonicalFile
+ // Copy the checkpoint to a temp dir to prevent changes to the original.
+ // Not doing this will lead to the test passing on the first run, but fail subsequent runs.
+ FileUtils.copyDirectory(new File(resourceUri), checkpointDir)
+
+ testStream(streamEvent) (
+ AddData(inputData, 1, 2, 3, 4, 5, 6),
+ StartStream(Trigger.Once, checkpointLocation = checkpointDir.getAbsolutePath),
+ ExpectFailure[IllegalStateException](e => {
Review comment:
Okay, thank you. I'll fix it.
--
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