[SPARK-23503][SS] Enforce sequencing of committed epochs for Continuous Execution #20936

spaced4ndy · 2018-03-29T14:46:41Z

What changes were proposed in this pull request?

Made changes to EpochCoordinator so that it enforces a commit order. In case a message for epoch n is lost and epoch (n + 1) is ready for commit before epoch n is, epoch (n + 1) will wait for epoch n to be committed first.

How was this patch tested?

Existing tests in ContinuousSuite and EpochCoordinatorSuite.

jose-torres

The general approach looks correct to me.

jose-torres · 2018-03-29T16:47:30Z

...re/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/EpochCoordinator.scala

+            partitionCommits.collect { case ((e, _), msg) if e == nextEpoch => msg }
+          logDebug(s"Committing epoch $nextEpoch.")
+          writer.commit(nextEpoch, nextEpochCommits.toArray)
+          query.commit(nextEpoch)


There's a bit of duplicated logic here - helper methods would probably be nice.

jose-torres · 2018-03-29T16:48:49Z

...re/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/EpochCoordinator.scala

+      } else {
+        logDebug(s"Epoch $epoch has received commits from all partitions" +
+          s"and is waiting for epoch ${epoch - 1} to be committed first.")
+        epochsWaitingToBeCommitted.add(epoch)


Maybe swap the order of the if else. I'd forgotten what the condition was for after scrolling down here.

spaced4ndy · 2018-04-08T13:20:39Z

@jose-torres Could you review latest changes for this PR please?

jose-torres · 2018-04-09T18:00:48Z

LGTM, once #20983 is committed and we can pull in the tests from that PR.

…nce committed epochs

spaced4ndy · 2018-04-18T10:26:31Z

@tdas @jose-torres Hi, I tested this with #20983, can we run jenkins tests now?

jose-torres · 2018-04-18T21:08:50Z

LGTM

jose-torres · 2018-04-18T21:14:05Z

...re/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/EpochCoordinator.scala

@@ -137,30 +137,65 @@ private[continuous] class EpochCoordinator(
  private val partitionOffsets =
    mutable.Map[(Long, Int), PartitionOffset]()

+  private var lastCommittedEpoch = startEpoch - 1
+  // Remembers epochs that have to wait for previous epochs to be committed first.
+  private val epochsWaitingToBeCommitted = mutable.HashSet.empty[Long]


This is orthogonal to the current PR, but I realized that both this and the commits/offsets maps are unbounded queues. We probably should introduce some SQLConf for the maximum epoch backlog, and report an error when too many stack up. I'll file a JIRA ticket for this.

felixcheung · 2018-04-19T07:00:41Z

Jenkins, ok to test

SparkQA · 2018-04-19T10:33:57Z

Test build #89555 has finished for PR 20936 at commit 478291b.

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

spaced4ndy · 2018-04-24T07:17:04Z

@felixcheung hi, could you merge this please? I'd use this to work on PR for epoch backlog issue Jose pointed out. Since it passes tests and Jose approves, I guess it's good to go.

As a side note, could you also add me to whitelist so that I don't waste maintainers' time to launch jenkins tests for future PRs? I don't know if it's usual for new contributors, but I see it's a possible work flow in Spark contributing guide.

tdas

This looks good, just requires a little bit more docs.

tdas · 2018-04-23T18:23:36Z

...re/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/EpochCoordinator.scala

      }
    }
  }

+  private def findCommitsForEpoch(epoch: Long): Iterable[WriterCommitMessage] = {


Add docs explaining what this does? As is, its hard to distinguish just from the name the difference between findCommitsForEpoch and commitEpoch. I think the term "commit" is overloaded here - commit in findCommitsForEpoch refers to per-partition commits, whereas commit in commitEpoch refers to committing the epoch to the offset log. May be its better to differentiate more clearly. commitEpoch and findPartitionCommitsForEpoch. And add docs to both methods also helps.

SparkQA · 2018-04-26T11:33:03Z

Test build #89876 has finished for PR 20936 at commit b1b9985.

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

tdas · 2018-05-05T08:00:37Z

jenkins retest this please.

SparkQA · 2018-05-05T09:56:43Z

Test build #90245 has finished for PR 20936 at commit b1b9985.

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

spaced4ndy · 2018-05-06T09:55:26Z

@tdas Hi, could you give me some advice on how to resolve this error? I'm not sure how this PR could cause this, especially considering it was passing tests before and failed jenkins test says "It is not a test it is a sbt.testing.SuiteSelector" in description.

jose-torres · 2018-05-06T14:47:23Z

retest this please

jose-torres · 2018-05-06T14:49:28Z

(I agree that your PR isn't responsible here, there's a known problem with that suite.)

SparkQA · 2018-05-06T18:26:27Z

Test build #90281 has finished for PR 20936 at commit b1b9985.

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

spaced4ndy · 2018-05-08T12:52:51Z

@tdas Hi, seems like it's good to go. Could you merge this when you have time?

tdas · 2018-05-16T17:07:02Z

Jenkins retest this please.

tdas · 2018-05-16T17:07:19Z

@efimpoberezkin I will merge it. Let me do another round of tests.

spaced4ndy · 2018-05-17T04:59:08Z

retest this please

spaced4ndy · 2018-05-17T05:08:02Z

@tdas can't start tests

tdas · 2018-05-17T22:30:48Z

jenkins retest this please.

SparkQA · 2018-05-18T01:13:13Z

Test build #90758 has finished for PR 20936 at commit b1b9985.

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

jose-torres · 2018-05-18T01:19:33Z

ugh that flaky kafka test. It's already reported, and I've been looking into it this week albeit with little luck.

jose-torres · 2018-05-18T01:19:38Z

retest this please

SparkQA · 2018-05-18T02:24:00Z

Test build #4181 has finished for PR 20936 at commit b1b9985.

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

SparkQA · 2018-05-18T04:52:23Z

Test build #90764 has finished for PR 20936 at commit b1b9985.

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

tdas · 2018-05-18T23:55:45Z

I merged this to master. Thanks. And sorry for the delay.

jose-torres reviewed Mar 29, 2018

View reviewed changes

Efim Poberezkin added 4 commits April 18, 2018 13:29

[SPARK-23503][Structured Streaming] Continuous Execution should seque…

5a3aec9

…nce committed epochs

Moved duplicated code to methods and swapped the order of if else

88035d0

Added missing spaces to debug messages

0a3409a

Unignored tests for EpochCoordinator sequencing of late epochs

478291b

jose-torres reviewed Apr 18, 2018

View reviewed changes

tdas approved these changes Apr 26, 2018

View reviewed changes

Improved readability

b1b9985

spaced4ndy changed the title ~~[SPARK-23503][Structured Streaming] Enforcing sequencing of committed epochs for Continuous Execution~~ [SPARK-23503][SS] Enforce sequencing of committed epochs for Continuous Execution Apr 26, 2018

asfgit closed this in 434d74e May 18, 2018

gaborgsomogyi mentioned this pull request Dec 10, 2018

[SPARK-24063][SS] Add maximum epoch queue threshold for ContinuousExecution #23156

Closed

uncleGen mentioned this pull request Apr 8, 2019

[SPARK-27355][SS] Make query execution more sensitive to epoch message late or lost #24283

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-23503][SS] Enforce sequencing of committed epochs for Continuous Execution #20936

[SPARK-23503][SS] Enforce sequencing of committed epochs for Continuous Execution #20936

spaced4ndy commented Mar 29, 2018 •

edited

Loading

jose-torres left a comment

jose-torres Mar 29, 2018

jose-torres Mar 29, 2018

spaced4ndy commented Apr 8, 2018

jose-torres commented Apr 9, 2018

spaced4ndy commented Apr 18, 2018

jose-torres commented Apr 18, 2018

jose-torres Apr 18, 2018

felixcheung commented Apr 19, 2018

SparkQA commented Apr 19, 2018

spaced4ndy commented Apr 24, 2018 •

edited

Loading

tdas left a comment

tdas Apr 23, 2018

spaced4ndy Apr 26, 2018

SparkQA commented Apr 26, 2018

tdas commented May 5, 2018

SparkQA commented May 5, 2018

spaced4ndy commented May 6, 2018 •

edited

Loading

jose-torres commented May 6, 2018

jose-torres commented May 6, 2018

SparkQA commented May 6, 2018

spaced4ndy commented May 8, 2018

tdas commented May 16, 2018

tdas commented May 16, 2018

spaced4ndy commented May 17, 2018

spaced4ndy commented May 17, 2018

tdas commented May 17, 2018

SparkQA commented May 18, 2018

jose-torres commented May 18, 2018

jose-torres commented May 18, 2018

SparkQA commented May 18, 2018

SparkQA commented May 18, 2018

tdas commented May 18, 2018

[SPARK-23503][SS] Enforce sequencing of committed epochs for Continuous Execution #20936

[SPARK-23503][SS] Enforce sequencing of committed epochs for Continuous Execution #20936

Conversation

spaced4ndy commented Mar 29, 2018 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

jose-torres left a comment

Choose a reason for hiding this comment

jose-torres Mar 29, 2018

Choose a reason for hiding this comment

jose-torres Mar 29, 2018

Choose a reason for hiding this comment

spaced4ndy commented Apr 8, 2018

jose-torres commented Apr 9, 2018

spaced4ndy commented Apr 18, 2018

jose-torres commented Apr 18, 2018

jose-torres Apr 18, 2018

Choose a reason for hiding this comment

felixcheung commented Apr 19, 2018

SparkQA commented Apr 19, 2018

spaced4ndy commented Apr 24, 2018 • edited Loading

tdas left a comment

Choose a reason for hiding this comment

tdas Apr 23, 2018

Choose a reason for hiding this comment

spaced4ndy Apr 26, 2018

Choose a reason for hiding this comment

SparkQA commented Apr 26, 2018

tdas commented May 5, 2018

SparkQA commented May 5, 2018

spaced4ndy commented May 6, 2018 • edited Loading

jose-torres commented May 6, 2018

jose-torres commented May 6, 2018

SparkQA commented May 6, 2018

spaced4ndy commented May 8, 2018

tdas commented May 16, 2018

tdas commented May 16, 2018

spaced4ndy commented May 17, 2018

spaced4ndy commented May 17, 2018

tdas commented May 17, 2018

SparkQA commented May 18, 2018

jose-torres commented May 18, 2018

jose-torres commented May 18, 2018

SparkQA commented May 18, 2018

SparkQA commented May 18, 2018

tdas commented May 18, 2018

spaced4ndy commented Mar 29, 2018 •

edited

Loading

spaced4ndy commented Apr 24, 2018 •

edited

Loading

spaced4ndy commented May 6, 2018 •

edited

Loading