[SPARK-15022][SPARK-15023][SQL][Streaming] Add support for testing against the `ProcessingTime(intervalMS > 0)` trigger and `ManualClock` #12797

lw-lin · 2016-04-30T02:02:52Z

What changes were proposed in this pull request?

Currently in StreamTest, we have a StartStream which will start a streaming query against trigger ProcessTime(intervalMS = 0) and SystemClock.

We also need to test cases against ProcessTime(intervalMS > 0), which often requires ManualClock.

This patch:

fixes an issue of ProcessingTimeExecutor, where for a batch it should run batchRunner only once but might run multiple times under certain conditions;
adds support for testing against the ProcessingTime(intervalMS > 0) trigger and ManualClock, by specifying them as fields for StartStream, and by adding an AdvanceManualClock action;
adds a test, which takes advantage of the new StartStream and AdvanceManualClock, to test against PR#[SPARK-14942] Reduce delay between batch construction and execution .

How was this patch tested?

N/A

lw-lin · 2016-04-30T02:03:26Z

This paragraph focuses on the `ProcessingTimeExecutor` might run `batchRunner` multiple times issue.

Current implementation might run `batchRunner` multiple times for a batch

Let's have intervalMs = 100, then in the current implementation, we'll get:

nextBatchTime(0) = 0
nextBatchTime(1) = 100
...
nextBatchTime(99) = 100
nextBatchTime(100) = 100
nextBatchTime(101) = 200
...
nextBatchTime(199) = 200
nextBatchTime(200) = 200

Because nextBatchTime(nextBatchTime(0)) is still 0, we might get the unexpected execution:

we run batchRunner()
now = 0, so nextBatchTime(with now = 0) is 0
so we run batchRunner()
now = 0, so nextBatchTime(with now = 0) is 0
so we run batchRunner()
...
until now advance to 1, then nextBatchTime(with now = 1) is 100, then we jump to the next batch

This patch's fix

After this patch, we'll get:

nextBatchTime(0) = 100     // this is the change
nextBatchTime(1) = 100
...
nextBatchTime(99) = 100
nextBatchTime(100) = 200   // this is the change
nextBatchTime(101) = 200
...
nextBatchTime(199) = 200
nextBatchTime(200) = 300   // this is the change

Because nextBatchTime(nextBatchTime(0)) is now 200, we'll get the expected execution:

we run batchRunner()
now = 0, so nextBatchTime(with now = 0) is 100
we wait until the next batch
we run batchRunner()
now = 100, so nextBatchTime(with now = 100) is 200
we wait until the next batch
we run batchRunner()
...

Why would this matter?

Generally in normal runtime environment, this issue is triggered only under extreme conditions where nextBatchTime is multiple of intervalMS and batchRunner() runs very fast so now does not change.

However, if we run tests against ManualClock, this issue is almost always triggered. This is because this time how fast or slow batchRunner() runs is irrelevant -- now always does not change unless we advance the clock manually. So although it's expected that batchRunner() runs only once each time we advance the clock, facts are batchRunner() might run many times. So I think we should fix this.

lw-lin · 2016-04-30T02:51:38Z

...re/src/test/scala/org/apache/spark/sql/execution/streaming/ProcessingTimeExecutorSuite.scala

+    testNextBatchTimeAgainstClock(new SystemClock)
+  }
+
+  test("nextBatchTime against ManualClock") {


Please note the ProcessingTimeExecutor issue would fail this test without this patch, but would pass with this patch.

lw-lin · 2016-04-30T03:09:47Z

@marmbrus @tdas @zsxwing would you mind taking a look? Thanks!

SparkQA · 2016-04-30T03:26:21Z

Test build #57391 has finished for PR 12797 at commit 9d80b15.

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

marmbrus · 2016-05-02T17:29:39Z

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

  }

-  /** Return the next multiple of intervalMs */
+  /** Return the next multiple of intervalMs


Nit: comment style is off, we use javadoc style

marmbrus · 2016-05-02T17:54:31Z

Some minor comments about code understandability, but overall this looks good. Thanks for working on this!

zsxwing · 2016-05-02T17:55:23Z

sql/core/src/test/scala/org/apache/spark/sql/StreamTest.scala


  /** Starts the stream, resuming if data has already been processed.  It must not be running. */
-  case object StartStream extends StreamAction
+  case class StartStream(trigger: Trigger = null, triggerClock: Clock = null) extends StreamAction


You can use the same default values of StreamExecution. Then you don't need to handle the null case.

This layer of nulls was intended to delegate the default values of StreamExecution into these tests, so that we don't have to set the same default values in many places and maintain their consistency. But since it seems very unlikely that we would change the default values, so I've removed the nulls layer and followed your comments.

Thanks!

zsxwing · 2016-05-03T18:12:39Z

Looks pretty good. @lw-lin could you address the comments and resolve the conflicts?

# Conflicts: # sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamE xecution.scala

zsxwing · 2016-05-04T00:04:20Z

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

+   * Returns the start time in milliseconds for the next batch interval, given the current time.
+   * Note that a batch interval is inclusive with respect to its start time, and thus calling
+   * `nextBatchTime` with the result of a previous call should return the next interval. (i.e. given
+   * an interval of `100 ms`, `nextBatchTime(nextBatchTime(0)) = 200` rather than `0`).


nit: nextBatchTime(nextBatchTime(0)) = 200 -> nextBatchTime(nextBatchTime(0)) = 100

nextBatchTime(0) = 100, so nextBatchTime(nextBatchTime(0)) = 200?

Oh, right. Sorry for the mistake.

zsxwing · 2016-05-04T00:12:09Z

LGTM pending tests.

lw-lin · 2016-05-04T00:14:53Z

@marmbrus @zsxwing thank you for the review! :-)

SparkQA · 2016-05-04T01:12:19Z

Test build #57691 has finished for PR 12797 at commit bc89962.

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

zsxwing · 2016-05-04T17:24:36Z

Merging to master / 2.0. Thanks again @lw-lin

…ainst the `ProcessingTime(intervalMS > 0)` trigger and `ManualClock` ## What changes were proposed in this pull request? Currently in `StreamTest`, we have a `StartStream` which will start a streaming query against trigger `ProcessTime(intervalMS = 0)` and `SystemClock`. We also need to test cases against `ProcessTime(intervalMS > 0)`, which often requires `ManualClock`. This patch: - fixes an issue of `ProcessingTimeExecutor`, where for a batch it should run `batchRunner` only once but might run multiple times under certain conditions; - adds support for testing against the `ProcessingTime(intervalMS > 0)` trigger and `AdvanceManualClock`, by specifying them as fields for `StartStream`, and by adding an `AdvanceClock` action; - adds a test, which takes advantage of the new `StartStream` and `AdvanceManualClock`, to test against [PR#[SPARK-14942] Reduce delay between batch construction and execution ](#12725). ## How was this patch tested? N/A Author: Liwei Lin <[email protected]> Closes #12797 from lw-lin/add-trigger-test-support. (cherry picked from commit e597ec6) Signed-off-by: Shixiong Zhu <[email protected]>

lw-lin added 2 commits April 30, 2016 09:32

fix an issue of nextBatchTime against ManualClock

90ed692

Add support for testing against ProcessingTime(intervalMS>0)

9d80b15

lw-lin reviewed Apr 30, 2016
View reviewed changes

lw-lin mentioned this pull request Apr 30, 2016

[SPARK-14942][SQL][Streaming] Reduce delay between batch construction and execution #12725

Closed

marmbrus reviewed May 2, 2016
View reviewed changes

zsxwing reviewed May 2, 2016
View reviewed changes

lw-lin changed the title ~~[SPARK-15022][SPARK-15023] Add support for testing against the ProcessingTime(intervalMS > 0) trigger and ManualClock~~ [SPARK-15022][SPARK-15023][SQL][Streaming] Add support for testing against the ProcessingTime(intervalMS > 0) trigger and ManualClock May 3, 2016

lw-lin added 2 commits May 4, 2016 07:16

Merge remote-tracking branch 'apache/master' into add-trigger-3

b63653a

# Conflicts: # sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamE xecution.scala

address comments

bc89962

zsxwing reviewed May 4, 2016
View reviewed changes

asfgit closed this in e597ec6 May 4, 2016

lw-lin deleted the add-trigger-test-support branch June 11, 2016 03:05

[SPARK-15022][SPARK-15023][SQL][Streaming] Add support for testing against the ProcessingTime(intervalMS > 0) trigger and ManualClock #12797

[SPARK-15022][SPARK-15023][SQL][Streaming] Add support for testing against the ProcessingTime(intervalMS > 0) trigger and ManualClock #12797

Uh oh!

Conversation

lw-lin commented Apr 30, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

lw-lin commented Apr 30, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

This paragraph focuses on the ProcessingTimeExecutor might run batchRunner multiple times issue.

Current implementation might run batchRunner multiple times for a batch

This patch's fix

Why would this matter?

Uh oh!

lw-lin Apr 30, 2016

Choose a reason for hiding this comment

Uh oh!

lw-lin commented Apr 30, 2016

Uh oh!

SparkQA commented Apr 30, 2016

Uh oh!

marmbrus May 2, 2016

Choose a reason for hiding this comment

Uh oh!

marmbrus commented May 2, 2016

Uh oh!

zsxwing May 2, 2016

Choose a reason for hiding this comment

Uh oh!

lw-lin May 4, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zsxwing commented May 3, 2016

Uh oh!

zsxwing May 4, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lw-lin May 4, 2016

Choose a reason for hiding this comment

Uh oh!

zsxwing May 4, 2016

Choose a reason for hiding this comment

Uh oh!

zsxwing commented May 4, 2016

Uh oh!

lw-lin commented May 4, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented May 4, 2016

Uh oh!

zsxwing commented May 4, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

[SPARK-15022][SPARK-15023][SQL][Streaming] Add support for testing against the `ProcessingTime(intervalMS > 0)` trigger and `ManualClock` #12797

[SPARK-15022][SPARK-15023][SQL][Streaming] Add support for testing against the `ProcessingTime(intervalMS > 0)` trigger and `ManualClock` #12797

lw-lin commented Apr 30, 2016 •

edited

Loading

lw-lin commented Apr 30, 2016 •

edited

Loading

This paragraph focuses on the `ProcessingTimeExecutor` might run `batchRunner` multiple times issue.

Current implementation might run `batchRunner` multiple times for a batch

lw-lin May 4, 2016 •

edited

Loading

zsxwing May 4, 2016 •

edited

Loading

lw-lin commented May 4, 2016 •

edited

Loading