SPARK-1407 drain event queue before stopping event logger

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -874,7 +874,6 @@ class SparkContext(
   /** Shut down the SparkContext. */
   def stop() {
     ui.stop()
-    eventLogger.foreach(_.stop())
     // Do this only if not stopped already - best case effort.
     // prevent NPE if stopped more than once.
     val dagSchedulerCopy = dagScheduler
@@ -884,6 +883,7 @@ class SparkContext(
       cleaner.foreach(_.stop())
       dagSchedulerCopy.stop()
       listenerBus.stop()
+      eventLogger.foreach(_.stop())
       taskScheduler = null
       // TODO: Cache.stop()?
       env.stop()

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

@@ -36,6 +36,7 @@ private[spark] class LiveListenerBus extends SparkListenerBus with Logging {
   private val eventQueue = new LinkedBlockingQueue[SparkListenerEvent](EVENT_QUEUE_CAPACITY)
   private var queueFullErrorMessageLogged = false
   private var started = false
+  private var sparkListenerBus: Option[Thread] = _
 
   /**
    * Start sending events to attached listeners.
@@ -49,7 +50,7 @@ private[spark] class LiveListenerBus extends SparkListenerBus with Logging {
       throw new IllegalStateException("Listener bus already started!")
     }
     started = true
-    new Thread("SparkListenerBus") {
+    sparkListenerBus = Some(new Thread("SparkListenerBus") {
       setDaemon(true)
       override def run() {
         while (true) {
@@ -61,7 +62,8 @@ private[spark] class LiveListenerBus extends SparkListenerBus with Logging {
           postToAll(event)
         }
       }
-    }.start()
+    })
+    sparkListenerBus.foreach(_.start())
   }
 
   def post(event: SparkListenerEvent) {
@@ -97,5 +99,6 @@ private[spark] class LiveListenerBus extends SparkListenerBus with Logging {
       throw new IllegalStateException("Attempted to stop a listener bus that has not yet started!")
     }
     post(SparkListenerShutdown)
+    sparkListenerBus.foreach(_.join())
   }
 }

examples/src/main/scala/org/apache/spark/examples/SparkHdfsLR.scala

@@ -73,6 +73,7 @@ object SparkHdfsLR {
     }
 
     println("Final w: " + w)
+    sc.stop()
     System.exit(0)
   }
 }