[SPARK-11761] Prevent the call to StreamingContext#stop() in the listener bus's thread #9741

tedyu · 2015-11-16T19:03:30Z

See discussion toward the tail of #9723
From @zsxwing :

The user should not call stop or other long-time work in a listener since it will block the listener thread, and prevent from stopping SparkContext/StreamingContext.

I cannot see an approach since we need to stop the listener bus's thread before stopping SparkContext/StreamingContext totally.

Proposed solution is to prevent the call to StreamingContext#stop() in the listener bus's thread.

tedyu · 2015-11-16T19:07:59Z

@tdas @zsxwing
Please take a look

SparkQA · 2015-11-16T19:19:45Z

Test build #46001 has finished for PR 9741 at commit bc40285.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-11-16T19:43:20Z

Test build #46003 has finished for PR 9741 at commit 8f583b9.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

andrewor14 · 2015-11-16T19:43:48Z

@tedyu please file a JIRA and put it in the title. I've reminded you many times in the past already.

tedyu · 2015-11-16T20:06:19Z

@andrewor14
Was pulled into a meeting when filing the JIRA.

Will pay attention next time.

SparkQA · 2015-11-16T22:11:04Z

Test build #46009 has finished for PR 9741 at commit abab461.

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

tedyu · 2015-11-17T21:10:09Z

@tdas @zsxwing
Any feedback ?

zsxwing · 2015-11-17T21:56:43Z

@tedyu could you add a unit test for this fix? Otherwise LGTM

tdas · 2015-11-17T23:58:29Z

yeah. I can merge this to 1.6 if you add a unit test.

tedyu · 2015-11-18T00:21:34Z

Please take a look at the test and see what should be improved.

zsxwing · 2015-11-18T00:29:20Z

streaming/src/test/scala/org/apache/spark/streaming/StreamingListenerSuite.scala

+class StreamingContextStoppingCollector(val ssc: StreamingContext) extends StreamingListener {
+  override def onOutputOperationStarted(
+      outputOperationStarted: StreamingListenerOutputOperationStarted): Unit = {
+      ssc.stop()


the listener bus will just log the exception. You can catch the exception here and use a field to store it. Then you can assert the exception in the test.

zsxwing · 2015-11-18T00:59:05Z

streaming/src/test/scala/org/apache/spark/streaming/StreamingListenerSuite.scala

+ * A StreamingListener that calls StreamingContext.stop().
+ */
+class StreamingContextStoppingCollector(val ssc: StreamingContext) extends StreamingListener {
+  var sparkExSeen = false


~~Could you add @volatile?~~ Oh, it's not necessary since you check it after stopping. There must be some memory barrier already.

SparkQA · 2015-11-18T01:06:59Z

Test build #46134 has finished for PR 9741 at commit 8ff4c61.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2015-11-18T01:26:51Z

LGTM

SparkQA · 2015-11-18T02:11:29Z

Test build #46125 has finished for PR 9741 at commit d67a133.

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

tdas · 2015-11-18T02:13:59Z

This an old commit. I will wait for the test on the latest commit to pass. Otherwise LGTM.

SparkQA · 2015-11-18T02:45:47Z

Test build #46132 has finished for PR 9741 at commit e0c6163.

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

SparkQA · 2015-11-18T02:48:57Z

Test build #46136 has finished for PR 9741 at commit 922bf0b.

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

tedyu · 2015-11-18T02:53:11Z

For both 46132 and 46136:

[info] StreamingListenerSuite:
[info] - batch info reporting (752 milliseconds)
[info] - receiver info reporting (369 milliseconds)
[info] - output operation reporting (318 milliseconds)
[info] - don't call ssc.stop in listener (962 milliseconds)
[info] - onBatchCompleted with successful batch (1 second, 12 milliseconds)

zsxwing · 2015-11-18T02:56:14Z

retest this please

tedyu · 2015-11-18T03:03:53Z

DirectKafkaStreamSuite passed locally:

Run starting. Expected test count is: 6
DirectKafkaStreamSuite:
 - basic stream receiving with multiple topics and smallest starting offset
- receiving from largest starting offset
- creating stream by offset
- offset recovery
- Direct Kafka stream report input information
- using rate controller
Run completed in 37 seconds, 464 milliseconds.

tedyu · 2015-11-18T03:15:15Z

ERROR: Timeout after 15 minutes
ERROR: Error fetching remote repo 'origin'
hudson.plugins.git.GitException: Failed to fetch from https://github.com/apache/spark.git
    at hudson.plugins.git.GitSCM.fetchFrom(GitSCM.java:763)

tedyu · 2015-11-18T03:15:25Z

Jenkins, retest this please

tedyu · 2015-11-18T04:11:43Z

DirectKafkaStreamSuite failed in maven Jenkins:

https://amplab.cs.berkeley.edu/jenkins/job/Spark-Master-Maven-with-YARN/HADOOP_PROFILE=hadoop-2.4,label=spark-test/4166/testReport/

SparkQA · 2015-11-18T05:29:46Z

Test build #46158 has finished for PR 9741 at commit 922bf0b.

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

tdas · 2015-11-18T06:47:05Z

I am merging this to master and 1.6. Thanks @tedyu!

…ener bus's thread See discussion toward the tail of #9723 From zsxwing : ``` The user should not call stop or other long-time work in a listener since it will block the listener thread, and prevent from stopping SparkContext/StreamingContext. I cannot see an approach since we need to stop the listener bus's thread before stopping SparkContext/StreamingContext totally. ``` Proposed solution is to prevent the call to StreamingContext#stop() in the listener bus's thread. Author: tedyu <yuzhihong@gmail.com> Closes #9741 from tedyu/master. (cherry picked from commit 446738e) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

andrewor14 · 2015-11-19T23:56:11Z

Should we do this in SparkContext as well? The user can still do sc.stop() on job end or something.

tedyu · 2015-11-20T00:04:45Z

I agree.
New PR coming

… bus's thread This is continuation of SPARK-11761 Andrew suggested adding this protection. See tail of #9741 Author: tedyu <yuzhihong@gmail.com> Closes #9852 from tedyu/master. (cherry picked from commit 8101254) Signed-off-by: Shixiong Zhu <shixiong@databricks.com>

… bus's thread This is continuation of SPARK-11761 Andrew suggested adding this protection. See tail of #9741 Author: tedyu <yuzhihong@gmail.com> Closes #9852 from tedyu/master.

… bus's thread This is continuation of SPARK-11761 Andrew suggested adding this protection. See tail of apache/spark#9741 Author: tedyu <yuzhihong@gmail.com> Closes #9852 from tedyu/master.

Prevent the call to StreamingContext#stop() in the listener bus's thread

bc40285

Address Scalastyle check failures

8f583b9

Address Scalastyle check

abab461

tedyu changed the title ~~Prevent the call to StreamingContext#stop() in the listener bus's thread~~ [SPARK-11761] Prevent the call to StreamingContext#stop() in the listener bus's thread Nov 16, 2015

Add test

d67a133

zsxwing reviewed Nov 18, 2015
View reviewed changes

Modify the new test

e0c6163

zsxwing reviewed Nov 18, 2015
View reviewed changes

tedyu added 3 commits November 17, 2015 17:01

Add volatile to sparkExSeen

8ff4c61

Remove unused variable

4ef7de2

Volatile

922bf0b

asfgit closed this in 446738e Nov 18, 2015

tedyu mentioned this pull request Nov 20, 2015

[SPARK-11872] Prevent the call to SparkContext#stop() in the listener bus's thread #9852

Closed

[SPARK-11761] Prevent the call to StreamingContext#stop() in the listener bus's thread #9741

[SPARK-11761] Prevent the call to StreamingContext#stop() in the listener bus's thread #9741

Uh oh!

Conversation

tedyu commented Nov 16, 2015

Uh oh!

tedyu commented Nov 16, 2015

Uh oh!

SparkQA commented Nov 16, 2015

Uh oh!

SparkQA commented Nov 16, 2015

Uh oh!

andrewor14 commented Nov 16, 2015

Uh oh!

tedyu commented Nov 16, 2015

Uh oh!

SparkQA commented Nov 16, 2015

Uh oh!

tedyu commented Nov 17, 2015

Uh oh!

zsxwing commented Nov 17, 2015

Uh oh!

tdas commented Nov 17, 2015

Uh oh!

tedyu commented Nov 18, 2015

Uh oh!

zsxwing Nov 18, 2015

Choose a reason for hiding this comment

Uh oh!

zsxwing Nov 18, 2015

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Nov 18, 2015

Uh oh!

zsxwing commented Nov 18, 2015

Uh oh!

SparkQA commented Nov 18, 2015

Uh oh!

tdas commented Nov 18, 2015

Uh oh!

SparkQA commented Nov 18, 2015

Uh oh!

SparkQA commented Nov 18, 2015

Uh oh!

tedyu commented Nov 18, 2015

Uh oh!

zsxwing commented Nov 18, 2015

Uh oh!

tedyu commented Nov 18, 2015

Uh oh!

tedyu commented Nov 18, 2015

Uh oh!

tedyu commented Nov 18, 2015

Uh oh!

tedyu commented Nov 18, 2015

Uh oh!

SparkQA commented Nov 18, 2015

Uh oh!

tdas commented Nov 18, 2015

Uh oh!

andrewor14 commented Nov 19, 2015

Uh oh!

tedyu commented Nov 20, 2015

Uh oh!

Uh oh!