[SPARK-15703] Make ListenerBus event queue configurable

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

@@ -249,7 +249,7 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
   def isStopped: Boolean = stopped.get()
 
   // An asynchronous listener bus for Spark events
-  private[spark] val listenerBus = new LiveListenerBus
+  private[spark] val listenerBus = new LiveListenerBus(this)
 
   // This function allows components created by SparkEnv to be mocked in unit tests:
   private[spark] def createSparkEnv(
@@ -2147,7 +2147,7 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
         }
     }
 
-    listenerBus.start(this)
+    listenerBus.start()
     _listenerBusStarted = true
   }
 

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

@@ -32,18 +32,17 @@ import org.apache.spark.util.Utils
  * has started will events be actually propagated to all attached listeners. This listener bus
  * is stopped when `stop()` is called, and it will drop further events after stopping.
  */
-private[spark] class LiveListenerBus extends SparkListenerBus {
+private[spark] class LiveListenerBus(val sparkContext: SparkContext) extends SparkListenerBus {
 
   self =>
 
   import LiveListenerBus._
 
-  private var sparkContext: SparkContext = null
-
   // Cap the capacity of the event queue so we get an explicit error (rather than
   // an OOM exception) if it's perpetually being added to more quickly than it's being drained.
-  private val EVENT_QUEUE_CAPACITY = 10000
-  private val eventQueue = new LinkedBlockingQueue[SparkListenerEvent](EVENT_QUEUE_CAPACITY)
+  private lazy val EVENT_QUEUE_CAPACITY = sparkContext.conf.
+    getInt("spark.scheduler.listenerbus.eventqueue.size", 10000)
+  private lazy val eventQueue = new LinkedBlockingQueue[SparkListenerEvent](EVENT_QUEUE_CAPACITY)
 
   // Indicate if `start()` is called
   private val started = new AtomicBoolean(false)
@@ -96,11 +95,9 @@ private[spark] class LiveListenerBus extends SparkListenerBus {
    * listens for any additional events asynchronously while the listener bus is still running.
    * This should only be called once.
    *
-   * @param sc Used to stop the SparkContext in case the listener thread dies.
    */
-  def start(sc: SparkContext): Unit = {
+  def start(): Unit = {
     if (started.compareAndSet(false, true)) {
-      sparkContext = sc
       listenerThread.start()
     } else {
       throw new IllegalStateException(s"$name already started!")

core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala

@@ -142,14 +142,14 @@ class EventLoggingListenerSuite extends SparkFunSuite with LocalSparkContext wit
     extraConf.foreach { case (k, v) => conf.set(k, v) }
     val logName = compressionCodec.map("test-" + _).getOrElse("test")
     val eventLogger = new EventLoggingListener(logName, None, testDirPath.toUri(), conf)
-    val listenerBus = new LiveListenerBus
+    val listenerBus = new LiveListenerBus(sc)
     val applicationStart = SparkListenerApplicationStart("Greatest App (N)ever", None,
       125L, "Mickey", None)
     val applicationEnd = SparkListenerApplicationEnd(1000L)
 
     // A comprehensive test on JSON de/serialization of all events is in JsonProtocolSuite
     eventLogger.start()
-    listenerBus.start(sc)
+    listenerBus.start()
     listenerBus.addListener(eventLogger)
     listenerBus.postToAll(applicationStart)
     listenerBus.postToAll(applicationEnd)

core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala

@@ -39,11 +39,11 @@ class SparkListenerSuite extends SparkFunSuite with LocalSparkContext with Match
   test("don't call sc.stop in listener") {
     sc = new SparkContext("local", "SparkListenerSuite")
     val listener = new SparkContextStoppingListener(sc)
-    val bus = new LiveListenerBus
+    val bus = new LiveListenerBus(sc)
     bus.addListener(listener)
 
     // Starting listener bus should flush all buffered events
-    bus.start(sc)
+    bus.start()
     bus.post(SparkListenerJobEnd(0, jobCompletionTime, JobSucceeded))
     bus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS)
 
@@ -53,15 +53,15 @@ class SparkListenerSuite extends SparkFunSuite with LocalSparkContext with Match
 
   test("basic creation and shutdown of LiveListenerBus") {
     val counter = new BasicJobCounter
-    val bus = new LiveListenerBus
+    val bus = new LiveListenerBus(sc)
     bus.addListener(counter)
 
     // Listener bus hasn't started yet, so posting events should not increment counter
     (1 to 5).foreach { _ => bus.post(SparkListenerJobEnd(0, jobCompletionTime, JobSucceeded)) }
     assert(counter.count === 0)
 
     // Starting listener bus should flush all buffered events
-    bus.start(sc)
+    bus.start()
     bus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS)
     assert(counter.count === 5)
 
@@ -72,14 +72,14 @@ class SparkListenerSuite extends SparkFunSuite with LocalSparkContext with Match
 
     // Listener bus must not be started twice
     intercept[IllegalStateException] {
-      val bus = new LiveListenerBus
-      bus.start(sc)
-      bus.start(sc)
+      val bus = new LiveListenerBus(sc)
+      bus.start()
+      bus.start()
     }
 
     // ... or stopped before starting
     intercept[IllegalStateException] {
-      val bus = new LiveListenerBus
+      val bus = new LiveListenerBus(sc)
       bus.stop()
     }
   }
@@ -107,11 +107,11 @@ class SparkListenerSuite extends SparkFunSuite with LocalSparkContext with Match
       }
     }
 
-    val bus = new LiveListenerBus
+    val bus = new LiveListenerBus(sc)
     val blockingListener = new BlockingListener
 
     bus.addListener(blockingListener)
-    bus.start(sc)
+    bus.start()
     bus.post(SparkListenerJobEnd(0, jobCompletionTime, JobSucceeded))
 
     listenerStarted.acquire()
@@ -353,13 +353,13 @@ class SparkListenerSuite extends SparkFunSuite with LocalSparkContext with Match
     val badListener = new BadListener
     val jobCounter1 = new BasicJobCounter
     val jobCounter2 = new BasicJobCounter
-    val bus = new LiveListenerBus
+    val bus = new LiveListenerBus(sc)
 
     // Propagate events to bad listener first
     bus.addListener(badListener)
     bus.addListener(jobCounter1)
     bus.addListener(jobCounter2)
-    bus.start(sc)
+    bus.start()
 
     // Post events to all listeners, and wait until the queue is drained
     (1 to 5).foreach { _ => bus.post(SparkListenerJobEnd(0, jobCompletionTime, JobSucceeded)) }

core/src/test/scala/org/apache/spark/storage/BlockManagerReplicationSuite.scala

@@ -92,7 +92,8 @@ class BlockManagerReplicationSuite extends SparkFunSuite with Matchers with Befo
     conf.set("spark.storage.cachedPeersTtl", "10")
 
     master = new BlockManagerMaster(rpcEnv.setupEndpoint("blockmanager",
-      new BlockManagerMasterEndpoint(rpcEnv, true, conf, new LiveListenerBus)), conf, true)
+      new BlockManagerMasterEndpoint(rpcEnv, true, conf,
+        new LiveListenerBus(new SparkContext(conf)))), conf, true)
     allStores.clear()
   }
 

core/src/test/scala/org/apache/spark/storage/BlockManagerSuite.scala

@@ -108,7 +108,8 @@ class BlockManagerSuite extends SparkFunSuite with Matchers with BeforeAndAfterE
     conf.set("spark.driver.port", rpcEnv.address.port.toString)
 
     master = new BlockManagerMaster(rpcEnv.setupEndpoint("blockmanager",
-      new BlockManagerMasterEndpoint(rpcEnv, true, conf, new LiveListenerBus)), conf, true)
+      new BlockManagerMasterEndpoint(rpcEnv, true, conf,
+        new LiveListenerBus(new SparkContext(conf)))), conf, true)
 
     val initialize = PrivateMethod[Unit]('initialize)
     SizeEstimator invokePrivate initialize()

core/src/test/scala/org/apache/spark/ui/storage/StorageTabSuite.scala

@@ -19,7 +19,7 @@ package org.apache.spark.ui.storage
 
 import org.scalatest.BeforeAndAfter
 
-import org.apache.spark.{SparkConf, SparkFunSuite, Success}
+import org.apache.spark.{SparkConf, SparkContext, SparkFunSuite, Success}
 import org.apache.spark.executor.TaskMetrics
 import org.apache.spark.scheduler._
 import org.apache.spark.storage._
@@ -43,8 +43,9 @@ class StorageTabSuite extends SparkFunSuite with BeforeAndAfter {
   private val bm1 = BlockManagerId("big", "dog", 1)
 
   before {
-    bus = new LiveListenerBus
-    storageStatusListener = new StorageStatusListener(new SparkConf())
+    val conf = new SparkConf()
+    bus = new LiveListenerBus(new SparkContext(conf))
+    storageStatusListener = new StorageStatusListener(conf)
     storageListener = new StorageListener(storageStatusListener)
     bus.addListener(storageStatusListener)
     bus.addListener(storageListener)

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

@@ -78,7 +78,8 @@ class ReceivedBlockHandlerSuite
     conf.set("spark.driver.port", rpcEnv.address.port.toString)
 
     blockManagerMaster = new BlockManagerMaster(rpcEnv.setupEndpoint("blockmanager",
-      new BlockManagerMasterEndpoint(rpcEnv, true, conf, new LiveListenerBus)), conf, true)
+      new BlockManagerMasterEndpoint(rpcEnv, true, conf,
+        new LiveListenerBus(new SparkContext(conf)))), conf, true)
 
     storageLevel = StorageLevel.MEMORY_ONLY_SER
     blockManager = createBlockManager(blockManagerSize, conf)