Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/4006#discussion_r23511814
--- Diff:
core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala ---
@@ -29,113 +28,18 @@ import org.apache.spark.util.Utils
* has started will events be actually propagated to all attached
listeners. This listener bus
* is stopped when it receives a SparkListenerShutdown event, which is
posted using stop().
*/
-private[spark] class LiveListenerBus extends SparkListenerBus with Logging
{
+private[spark] class LiveListenerBus
+ extends ListenerBus[SparkListener,
SparkListenerEvent]("SparkListenerBus") with SparkListenerBus {
- /* Cap the capacity of the SparkListenerEvent 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 var queueFullErrorMessageLogged = false
- private var started = false
+ private val logDroppedEvent = new AtomicBoolean(false)
- // A counter that represents the number of events produced and consumed
in the queue
- private val eventLock = new Semaphore(0)
-
- private val listenerThread = new Thread("SparkListenerBus") {
- setDaemon(true)
- override def run(): Unit = Utils.logUncaughtExceptions {
- while (true) {
- eventLock.acquire()
- // Atomically remove and process this event
- LiveListenerBus.this.synchronized {
- val event = eventQueue.poll
- if (event == SparkListenerShutdown) {
- // Get out of the while loop and shutdown the daemon thread
- return
- }
- Option(event).foreach(postToAll)
- }
- }
+ override def onDropEvent(event: SparkListenerEvent): Unit = {
+ if (logDroppedEvent.compareAndSet(false, true)) {
+ // Only log the following message once to avoid duplicated annoying
logs.
+ 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.")
--- End diff --
you'll need a space after `with` on L40
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
