[SPARK-11572] Process outstanding requests after seeing stop flag

[tedyu]

This is to address the regression Josh discovered.

[SparkQA]

Test build #45961 has finished for PR 9723 at commit e208c5f.

• This patch fails to build.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #45966 has finished for PR 9723 at commit 3df8051.

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

[JoshRosen]

I'd like @zsxwing to help take a look at this and figure out whether this is a sufficient fix or whether we should revert your original patch and go back to the drawing board.

[JoshRosen]

It's possible that there's a cleaner way to fix the original bug that motivated the incorrect patch vs. trying to patch up the problems introduced by the fix by layering more fixes on top. Let's see.

[SparkQA]

Test build #45962 has finished for PR 9723 at commit cb4132d.

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

[SparkQA]

Test build #45968 has finished for PR 9723 at commit 8be6a7c.

• This patch fails to build.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #45973 has finished for PR 9723 at commit cd9b2f2.

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

[tedyu]

sbt.ForkMain$ForkError: Test failed with compression using codec org.apache.spark.io.LZ4CompressionCodec:

assertion failed: expected groupByKey to spill, but did not
    at scala.Predef$.assert(Predef.scala:179)
    at org.apache.spark.TestUtils$.assertSpilled(TestUtils.scala:169)
    at org.apache.spark.util.collection.ExternalAppendOnlyMapSuite.org$apache$spark$util$collection$ExternalAppendOnlyMapSuite$$testSimpleSpilling(ExternalAppendOnlyMapSuite.scala:254)
    at org.apache.spark.util.collection.ExternalAppendOnlyMapSuite$$anonfun$10$$anonfun$apply$mcV$sp$8.apply(ExternalAppendOnlyMapSuite.scala:218)
    at org.apache.spark.util.collection.ExternalAppendOnlyMapSuite$$anonfun$10$$anonfun$apply$mcV$sp$8.apply(ExternalAppendOnlyMapSuite.scala:216)
    at scala.collection.immutable.Stream.foreach(Stream.scala:547)

I think the above was not related to change in AsynchronousListenerBus

[zsxwing]

IMO, we should revert 3e0a6cf

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.

[JoshRosen]

I'm going to revert those commits now to try to unbreak the SBT builds.

[tedyu]

Sounds like we should prevent the call to StreamingContext#stop() in the listener bus's thread.
How about setting a ThreadLocal Boolean to indicate to StreamingContext that the stop() call should not be honored ?

[zsxwing]

Sounds like we should prevent the call to StreamingContext#stop() in the listener bus's thread.
How about setting a ThreadLocal Boolean to indicate to StreamingContext that the stop() call should not be honored ?

Sound like a good idea. Ping @tdas, what do you think?