Do not block on posting SparkListenerApplicationEnd

witgo · Apr 10, 2014 · 69d1b41 · 69d1b41
1 parent 19d5dd0
commit 69d1b41
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 13 deletions.
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -1188,7 +1188,7 @@ class SparkContext(config: SparkConf) extends Logging {
    * condition exists in which the listeners may stop before this event has been propagated.
    */
   private def postApplicationEnd() {
-    listenerBus.post(SparkListenerApplicationEnd(System.currentTimeMillis), blocking = true)
+    listenerBus.post(SparkListenerApplicationEnd(System.currentTimeMillis))
   }
 
   /** Post the environment update event once the task scheduler is ready */

diff --git a/core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala b/core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala
@@ -68,18 +68,13 @@ private[spark] class LiveListenerBus extends SparkListenerBus with Logging {
     started = true
   }
 
-  def post(event: SparkListenerEvent, blocking: Boolean = false) {
-    if (!blocking) {
-      val eventAdded = eventQueue.offer(event)
-      if (!eventAdded && !queueFullErrorMessageLogged) {
-        logError("Dropping SparkListenerEvent because no remaining room in event queue. " +
-          "This likely means one of the SparkListeners is too slow and cannot keep up with the " +
-          "rate at which tasks are being started by the scheduler.")
-        queueFullErrorMessageLogged = true
-      }
-    } else {
-      // Bypass the event queue and post to all attached listeners immediately
-      postToAll(event)
+  def post(event: SparkListenerEvent) {
+    val eventAdded = eventQueue.offer(event)
+    if (!eventAdded && !queueFullErrorMessageLogged) {
+      logError("Dropping SparkListenerEvent because no remaining room in event queue. " +
+        "This likely means one of the SparkListeners is too slow and cannot keep up with the " +
+        "rate at which tasks are being started by the scheduler.")
+      queueFullErrorMessageLogged = true
     }
   }