[SPARK-26425][SS] Add more constraint checks to avoid checkpoint corruption #25965

HeartSaVioR · 2019-09-29T15:44:37Z

What changes were proposed in this pull request?

Credits to @tdas who reported and described the fix to SPARK-26425. I just followed the description of the issue.

This patch adds more checks on commit log as well as file streaming source so that multiple concurrent runs of streaming query don't mess up the status of query/checkpoint. This patch addresses two different spots which are having a bit different issues:

FileStreamSource.fetchMaxOffset()

In structured streaming, we don't allow multiple streaming queries to run with same checkpoint (including concurrent runs of same query), so query should fail if it fails to write the metadata of specific batch ID due to same batch ID being written by others.

commit log

As described in JIRA issue, assertion is already applied to the offsetLog for the same reason.

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala

Lines 394 to 402 in 8167714

    
           if (shouldConstructNextBatch) { 
        
             // Commit the next batch offset range to the offset log 
        
             updateStatusMessage("Writing offsets to log") 
        
             reportTimeTaken("walCommit") { 
        
               assert(offsetLog.add(currentBatchId, 
        
                 availableOffsets.toOffsetSeq(sources, offsetSeqMetadata)), 
        
                 s"Concurrent update to the log. Multiple streaming jobs detected for $currentBatchId") 
        
               logInfo(s"Committed offsets for batch $currentBatchId. " + 
        
                 s"Metadata ${offsetSeqMetadata.toString}")

This patch applied the same for commit log.

Why are the changes needed?

This prevents the inconsistent behavior on streaming query and lets query fail instead.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

N/A, as the change is simple and obvious, and it's really hard to artificially reproduce the issue.

HeartSaVioR · 2019-09-29T15:47:19Z

I'm also feeling that commitLog can be also guarded for same, but want to be sure before updating the patch.

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala

Lines 566 to 572 in 8167714

    
           withProgressLocked { 
        
             sinkCommitProgress = batchSinkProgress 
        
             watermarkTracker.updateWatermark(lastExecution.executedPlan) 
        
             commitLog.add(currentBatchId, CommitMetadata(watermarkTracker.currentWatermark)) 
        
             committedOffsets ++= availableOffsets 
        
           } 
        
           logDebug(s"Completed batch ${currentBatchId}")

HeartSaVioR · 2019-09-29T15:49:16Z

cc. @tdas @zsxwing @jose-torres @gaborgsomogyi

SparkQA · 2019-09-29T19:21:03Z

Test build #111570 has finished for PR 25965 at commit 726d920.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2019-12-06T12:45:25Z

@tdas @zsxwing @jose-torres @gaborgsomogyi Kindly reminder.

HeartSaVioR · 2020-01-13T02:58:31Z

retest this, please

SparkQA · 2020-01-13T07:27:45Z

Test build #116587 has finished for PR 25965 at commit 726d920.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-04-13T22:50:12Z

retest this, please

SparkQA · 2020-04-14T04:30:01Z

Test build #121235 has finished for PR 25965 at commit 726d920.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-04-30T12:46:47Z

retest this, please

SparkQA · 2020-04-30T18:19:13Z

Test build #122134 has finished for PR 25965 at commit 726d920.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-05-21T18:50:02Z

Test build #122934 has finished for PR 25965 at commit 1744179.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

… to avoid checkpoint corruption

HeartSaVioR · 2020-05-26T22:36:27Z

Rebased - the part of this PR was resolved via SPARK-30915. I've updated to address commit log instead.

SparkQA · 2020-05-27T01:38:09Z

Test build #123147 has finished for PR 25965 at commit d15acef.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-05-27T02:16:04Z

retest this, please

SparkQA · 2020-05-27T05:32:34Z

Test build #123157 has finished for PR 25965 at commit d15acef.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-05-28T03:46:45Z

retest this, please

SparkQA · 2020-05-28T07:05:01Z

Test build #123211 has finished for PR 25965 at commit d15acef.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-05-28T07:09:08Z

retest this, please

SparkQA · 2020-05-28T11:44:35Z

Test build #123221 has finished for PR 25965 at commit d15acef.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-06-14T01:48:53Z

retest this, please

SparkQA · 2020-06-14T06:58:51Z

Test build #123992 has finished for PR 25965 at commit d15acef.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-06-14T07:17:41Z

 pyspark.mllib.tests.test_streaming_algorithms.StreamingLinearRegressionWithTests.test_train_prediction

 Error Details

1.672640157855923 not greater than 2

 Stack Trace

Traceback (most recent call last):
  File "/home/jenkins/workspace/SparkPullRequestBuilder@3/python/pyspark/mllib/tests/test_streaming_algorithms.py", line 466, in test_train_prediction
    eventually(condition, timeout=180.0)
  File "/home/jenkins/workspace/SparkPullRequestBuilder@3/python/pyspark/testing/utils.py", line 81, in eventually
    lastValue = condition()
  File "/home/jenkins/workspace/SparkPullRequestBuilder@3/python/pyspark/mllib/tests/test_streaming_algorithms.py", line 461, in condition
    self.assertGreater(errors[1] - errors[-1], 2)
AssertionError: 1.672640157855923 not greater than 2

Known flaky test.

HeartSaVioR · 2020-06-14T07:17:52Z

retest this, please

SparkQA · 2020-06-14T12:10:00Z

Test build #124003 has finished for PR 25965 at commit d15acef.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-07-12T20:28:38Z

retest this, please

SparkQA · 2020-07-13T02:16:42Z

Test build #125731 has finished for PR 25965 at commit d15acef.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-08-18T05:02:00Z

retest this, please

SparkQA · 2020-08-18T07:05:03Z

Test build #127538 has finished for PR 25965 at commit d15acef.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-08-18T07:08:50Z

retest this, please

SparkQA · 2020-08-18T08:04:00Z

Test build #127553 has finished for PR 25965 at commit d15acef.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-08-18T08:12:21Z

retest this, please

SparkQA · 2020-08-18T13:32:07Z

Test build #127561 has finished for PR 25965 at commit d15acef.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-09-15T00:23:11Z

retest this, please

SparkQA · 2020-09-15T07:05:01Z

Test build #128679 has finished for PR 25965 at commit d15acef.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-09-15T07:27:37Z

retest this, please

SparkQA · 2020-09-15T12:08:11Z

Test build #128704 has finished for PR 25965 at commit d15acef.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

viirya

Looks good. This change is simple and straightforward.

viirya · 2020-09-16T05:56:54Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala

      watermarkTracker.updateWatermark(lastExecution.executedPlan)
-      commitLog.add(currentBatchId, CommitMetadata(watermarkTracker.currentWatermark))
+      assert(commitLog.add(currentBatchId, CommitMetadata(watermarkTracker.currentWatermark)),
+        s"Concurrent update to the log. Multiple streaming jobs detected for $currentBatchId")


update to the log -> update to the commit log?

Good point. Addressed.

SparkQA · 2020-09-16T07:05:02Z

Test build #128745 has finished for PR 25965 at commit a885212.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-09-16T07:23:30Z

retest this, please

SparkQA · 2020-09-16T12:15:05Z

Test build #128753 has finished for PR 25965 at commit a885212.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2020-09-16T16:01:44Z

Again, this is simple and straightforward change. Tests are passed. If no objection, I think we can move forward to merge this.

HeartSaVioR · 2020-09-16T23:59:24Z

Thanks for reviewing. This PR didn't have any valid review comments in 1 year despite of mentioning, so I don't expect more reviews. I'll go ahead merging.

HeartSaVioR · 2020-09-17T00:02:42Z

Merged into master branch.

dongjoon-hyun added the STRUCTURED STREAMING label Sep 29, 2019

HeartSaVioR mentioned this pull request May 13, 2020

[SPARK-30915][SS] CompactibleFileStreamLog: Avoid reading the metadata log file when finding the latest batch ID #27664

Closed

probot-autolabeler bot added the SQL label May 21, 2020

[SPARK-26425][SS] Add more constraint checks in file streaming source…

d15acef

… to avoid checkpoint corruption

HeartSaVioR force-pushed the SPARK-26425 branch from 1744179 to d15acef Compare May 26, 2020 22:23

HeartSaVioR changed the title ~~[SPARK-26425][SS] Add more constraint checks in file streaming source to avoid checkpoint corruption~~ [SPARK-26425][SS] Add more constraint checks to avoid checkpoint corruption May 26, 2020

viirya approved these changes Sep 16, 2020

View reviewed changes

Reflect review comment

a885212

HeartSaVioR closed this in d936cb3 Sep 17, 2020

HeartSaVioR deleted the SPARK-26425 branch September 17, 2020 00:02

	if (shouldConstructNextBatch) {
	// Commit the next batch offset range to the offset log
	updateStatusMessage("Writing offsets to log")
	reportTimeTaken("walCommit") {
	assert(offsetLog.add(currentBatchId,
	availableOffsets.toOffsetSeq(sources, offsetSeqMetadata)),
	s"Concurrent update to the log. Multiple streaming jobs detected for $currentBatchId")
	logInfo(s"Committed offsets for batch $currentBatchId. " +
	s"Metadata ${offsetSeqMetadata.toString}")

[SPARK-26425][SS] Add more constraint checks to avoid checkpoint corruption #25965

[SPARK-26425][SS] Add more constraint checks to avoid checkpoint corruption #25965

Uh oh!

Conversation

HeartSaVioR commented Sep 29, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

HeartSaVioR commented Sep 29, 2019

Uh oh!

HeartSaVioR commented Sep 29, 2019

Uh oh!

SparkQA commented Sep 29, 2019

Uh oh!

HeartSaVioR commented Dec 6, 2019

Uh oh!

HeartSaVioR commented Jan 13, 2020

Uh oh!

SparkQA commented Jan 13, 2020

Uh oh!

HeartSaVioR commented Apr 13, 2020

Uh oh!

SparkQA commented Apr 14, 2020

Uh oh!

HeartSaVioR commented Apr 30, 2020

Uh oh!

SparkQA commented Apr 30, 2020

Uh oh!

SparkQA commented May 21, 2020

Uh oh!

HeartSaVioR commented May 26, 2020

Uh oh!

SparkQA commented May 27, 2020

Uh oh!

HeartSaVioR commented May 27, 2020

Uh oh!

SparkQA commented May 27, 2020

Uh oh!

HeartSaVioR commented May 28, 2020

Uh oh!

SparkQA commented May 28, 2020

Uh oh!

HeartSaVioR commented May 28, 2020

Uh oh!

SparkQA commented May 28, 2020

Uh oh!

HeartSaVioR commented Jun 14, 2020

Uh oh!

SparkQA commented Jun 14, 2020

Uh oh!

HeartSaVioR commented Jun 14, 2020

Uh oh!

HeartSaVioR commented Jun 14, 2020

Uh oh!

SparkQA commented Jun 14, 2020

Uh oh!

HeartSaVioR commented Jul 12, 2020

Uh oh!

SparkQA commented Jul 13, 2020

Uh oh!

HeartSaVioR commented Aug 18, 2020

Uh oh!

SparkQA commented Aug 18, 2020

Uh oh!

HeartSaVioR commented Aug 18, 2020

Uh oh!

SparkQA commented Aug 18, 2020

Uh oh!

HeartSaVioR commented Aug 18, 2020

Uh oh!

SparkQA commented Aug 18, 2020

Uh oh!

HeartSaVioR commented Sep 15, 2020

Uh oh!

SparkQA commented Sep 15, 2020

Uh oh!

HeartSaVioR commented Sep 15, 2020

HeartSaVioR commented Sep 29, 2019 •

edited

Loading