[SPARK-28574][CORE] Allow to config different sizes for event queues

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

@@ -46,8 +46,13 @@ private class AsyncEventQueue(
 
   // Cap the capacity of the 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 eventQueue = new LinkedBlockingQueue[SparkListenerEvent](
-    conf.get(LISTENER_BUS_EVENT_QUEUE_CAPACITY))
+  // The capacity can be configured by spark.scheduler.listenerbus.eventqueue.${name}.capacity,
+  // if no such conf is specified, use the value specified in
+  // LISTENER_BUS_EVENT_QUEUE_CAPACITY
+  protected def capacity: Int = conf.getInt(
+    s"spark.scheduler.listenerbus.eventqueue.${name}.capacity",
+     conf.get(LISTENER_BUS_EVENT_QUEUE_CAPACITY))
+  private val eventQueue = new LinkedBlockingQueue[SparkListenerEvent](capacity)
 
   // Keep the event count separately, so that waitUntilEmpty() can be implemented properly;
   // this allows that method to return only when the events in the queue have been fully

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

@@ -236,6 +236,13 @@ private[spark] class LiveListenerBus(conf: SparkConf) {
     queues.asScala.map(_.name).toSet
   }
 
+  // For testing only.
+  private[scheduler] def getQueueCapacity(name: String): Int = {
+    queues.asScala.find(_.name == name) match {
+      case Some(queue) => queue.capacity
+      case None => -1
+    }
+  }
 }
 
 private[spark] object LiveListenerBus {

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

@@ -532,6 +532,28 @@ class SparkListenerSuite extends SparkFunSuite with LocalSparkContext with Match
     }
   }
 
+  test("event queue size can be configued through spark conf") {
+    val conf = new SparkConf(false)
+      .set(LISTENER_BUS_EVENT_QUEUE_CAPACITY, 5)
+      .set("spark.scheduler.listenerbus.eventqueue.shared.capacity", "1")
+      .set("spark.scheduler.listenerbus.eventqueue.eventLog.capacity", "2")
+
+    val bus = new LiveListenerBus(conf)
+    val counter1 = new BasicJobCounter()
+    val counter2 = new BasicJobCounter()
+    val counter3 = new BasicJobCounter()
+
+    bus.addToSharedQueue(counter1)
+    bus.addToStatusQueue(counter2)
+    bus.addToEventLogQueue(counter3)
+
+    assert(bus.activeQueues() === Set(SHARED_QUEUE, APP_STATUS_QUEUE, EVENT_LOG_QUEUE))
+    // check the size of each queue
+    assert(bus.getQueueCapacity(SHARED_QUEUE) == 1)
+    assert(bus.getQueueCapacity(APP_STATUS_QUEUE) == 5)
+    assert(bus.getQueueCapacity(EVENT_LOG_QUEUE) == 2)
+  }
+
   /**
    * Assert that the given list of numbers has an average that is greater than zero.
    */