Repository: spark Updated Branches: refs/heads/branch-1.6 eced2766b -> 053c63ecf
Revert "[SPARK-11572] Exit AsynchronousListenerBus thread when stop() is called" This reverts commit 93ac30741cd0ed99512f01525fbda8a08c87967a. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/053c63ec Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/053c63ec Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/053c63ec Branch: refs/heads/branch-1.6 Commit: 053c63ecfec63b32d27de74409251d6647210055 Parents: eced276 Author: Josh Rosen <[email protected]> Authored: Sun Nov 15 22:39:45 2015 -0800 Committer: Josh Rosen <[email protected]> Committed: Sun Nov 15 22:39:45 2015 -0800 ---------------------------------------------------------------------- .../org/apache/spark/util/AsynchronousListenerBus.scala | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/spark/blob/053c63ec/core/src/main/scala/org/apache/spark/util/AsynchronousListenerBus.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/org/apache/spark/util/AsynchronousListenerBus.scala b/core/src/main/scala/org/apache/spark/util/AsynchronousListenerBus.scala index b3b54af..c20627b 100644 --- a/core/src/main/scala/org/apache/spark/util/AsynchronousListenerBus.scala +++ b/core/src/main/scala/org/apache/spark/util/AsynchronousListenerBus.scala @@ -66,12 +66,15 @@ private[spark] abstract class AsynchronousListenerBus[L <: AnyRef, E](name: Stri processingEvent = true } try { - if (stopped.get()) { + val event = eventQueue.poll + if (event == null) { // Get out of the while loop and shutdown the daemon thread + if (!stopped.get) { + throw new IllegalStateException("Polling `null` from eventQueue means" + + " the listener bus has been stopped. So `stopped` must be true") + } return } - val event = eventQueue.poll - assert(event != null, "event queue was empty but the listener bus was not stopped") postToAll(event) } finally { self.synchronized { --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
