[SPARK-25399][SS] Continuous processing state should not affect microbatch execution jobs #22386

mukulmurthy · 2018-09-10T23:57:10Z

What changes were proposed in this pull request?

The leftover state from running a continuous processing streaming job should not affect later microbatch execution jobs. If a continuous processing job runs and the same thread gets reused for a microbatch execution job in the same environment, the microbatch job could get wrong answers because it can attempt to load the wrong version of the state.

How was this patch tested?

New and existing unit tests

…ssing

…ariable

mukulmurthy · 2018-09-10T23:57:32Z

@tdas and @jose-torres for review

SparkQA · 2018-09-11T03:39:16Z

Test build #95908 has finished for PR 22386 at commit c2f813b.

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

xuanyuanking · 2018-09-11T15:57:11Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala

+
+  test("is_continuous_processing property should be true for continuous processing") {
+    val input = ContinuousMemoryStream[Int]
+    var x: String = ""


xuanyuanking · 2018-09-11T16:09:52Z

If a continuous processing job runs and the same thread gets reused for a microbatch execution job in the same environment

Little confuse about this scenario, could you explain more? I mean its only happened in UT or we may meet this on product env?

jose-torres

LGTM other than 1 nit + the comment from @xuanyuanking

jose-torres · 2018-09-11T16:12:27Z

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

 }

 object ContinuousExecution {
+  val IS_CONTINUOUS_PROCESSING = "__is_continuous_processing"


nit: I think this belongs in StreamExecution, since both ContinuousExecution and MicroBatchExecution set it.

xuanyuanking · 2018-09-11T16:16:16Z

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

-    val currentVersion = EpochTracker.getCurrentEpoch match {
-      case None => storeVersion
-      case Some(value) => value
+    val isContinuous = Option(ctxt.getLocalProperty(ContinuousExecution.IS_CONTINUOUS_PROCESSING))


Just simple toBoolean here is OK？Cause you set default value both MicroBatch and Continuous side.

I think I'd rather keep it as is to be more resilient for the future.

mukulmurthy · 2018-09-11T17:11:39Z

If a continuous processing job runs and the same thread gets reused for a microbatch execution job in the same environment
Little confuse about this scenario, could you explain more? I mean its only happened in UT or we may meet this on product env?

@xuanyuanking It theoretically could have been encountered in production, but continuous processing is considered an experimental feature. The only way to encounter it in production is to run a continuous processing stream and then a microbatch stream in the same spark cluster and have an execution thread get reused. The bug is in StateStoreRDD; EpochTracker sets a ThreadLocal variable called currentEpoch and StateStoreRDD checks for the existence of this variable to decide if the current streaming job is continuous or microbatch.

tdas · 2018-09-11T18:52:18Z

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

-      case Some(value) => value
+    val isContinuous = Option(ctxt.getLocalProperty(StreamExecution.IS_CONTINUOUS_PROCESSING))
+      .map(_.toBoolean)
+    val currentVersion = if (isContinuous.contains(true)) {


super nit: this looks weird. rather i would do change the previous line val isContinuous = ... .map(_.toBoolean).getOrElse(false)

tdas · 2018-09-11T18:53:50Z

LGTM. Just one super nit.

SparkQA · 2018-09-11T19:22:49Z

Test build #95959 has finished for PR 22386 at commit 3ebbed3.

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

SparkQA · 2018-09-11T22:45:07Z

Test build #95961 has finished for PR 22386 at commit 4d4beef.

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

…batch execution jobs ## What changes were proposed in this pull request? The leftover state from running a continuous processing streaming job should not affect later microbatch execution jobs. If a continuous processing job runs and the same thread gets reused for a microbatch execution job in the same environment, the microbatch job could get wrong answers because it can attempt to load the wrong version of the state. ## How was this patch tested? New and existing unit tests Closes #22386 from mukulmurthy/25399-streamthread. Authored-by: Mukul Murthy <[email protected]> Signed-off-by: Tathagata Das <[email protected]> (cherry picked from commit 9f5c5b4) Signed-off-by: Tathagata Das <[email protected]>

xuanyuanking · 2018-09-12T01:56:35Z

Great thanks for your comment and fix @mukulmurthy! We'll also port this to our folk.

…batch execution jobs ## What changes were proposed in this pull request? The leftover state from running a continuous processing streaming job should not affect later microbatch execution jobs. If a continuous processing job runs and the same thread gets reused for a microbatch execution job in the same environment, the microbatch job could get wrong answers because it can attempt to load the wrong version of the state. ## How was this patch tested? New and existing unit tests Closes apache#22386 from mukulmurthy/25399-streamthread. Authored-by: Mukul Murthy <[email protected]> Signed-off-by: Tathagata Das <[email protected]>

mukulmurthy added 4 commits September 10, 2018 14:40

Set a task context property to indicate we are using continuous proce…

f91f365

…ssing

Use property to decide which version instaed of reading ThreadLocal v…

2b86b2f

…ariable

Set is continuous processing property to false for microbatch

0cf0751

tests

c2f813b

xuanyuanking reviewed Sep 11, 2018

View reviewed changes

jose-torres reviewed Sep 11, 2018

View reviewed changes

xuanyuanking reviewed Sep 11, 2018

View reviewed changes

address PR comments

3ebbed3

tdas reviewed Sep 11, 2018

View reviewed changes

mukulmurthy changed the title ~~[SPARK-25399] Continuous processing state should not affect microbatch execution jobs~~ [SPARK-25399][SS] Continuous processing state should not affect microbatch execution jobs Sep 11, 2018

TD's comment

4d4beef

asfgit closed this in 9f5c5b4 Sep 11, 2018

mukulmurthy deleted the 25399-streamthread branch September 17, 2018 21:45

[SPARK-25399][SS] Continuous processing state should not affect microbatch execution jobs #22386

[SPARK-25399][SS] Continuous processing state should not affect microbatch execution jobs #22386

Uh oh!

Conversation

mukulmurthy commented Sep 10, 2018

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

mukulmurthy commented Sep 10, 2018

Uh oh!

SparkQA commented Sep 11, 2018

Uh oh!

xuanyuanking Sep 11, 2018

Choose a reason for hiding this comment

Uh oh!

xuanyuanking commented Sep 11, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jose-torres left a comment

Choose a reason for hiding this comment

Uh oh!

jose-torres Sep 11, 2018

Choose a reason for hiding this comment

Uh oh!

xuanyuanking Sep 11, 2018

Choose a reason for hiding this comment

Uh oh!

mukulmurthy Sep 11, 2018

Choose a reason for hiding this comment

Uh oh!

mukulmurthy commented Sep 11, 2018

Uh oh!

tdas Sep 11, 2018

Choose a reason for hiding this comment

Uh oh!

tdas commented Sep 11, 2018

Uh oh!

SparkQA commented Sep 11, 2018

Uh oh!

SparkQA commented Sep 11, 2018

Uh oh!

xuanyuanking commented Sep 12, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

xuanyuanking commented Sep 11, 2018 •

edited

Loading

xuanyuanking commented Sep 12, 2018 •

edited

Loading