[SPARK-27804] LiveListenerBus#addToQueue : create multiple AsyncEventQueues under race condition

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Not A Problem
• Affects Version/s: 2.4.3
• Component/s: Spark Core
• Assignee: Unassigned
• Reporter: duyu
• Created: 22/May/19 11:37
• Updated: 22/May/19 19:09
• Resolved: 22/May/19 18:39

Description

LiveListenerBus.scala

  private[spark] def addToQueue(
      listener: SparkListenerInterface,
      queue: String): Unit = synchronized {
    if (stopped.get()) {
      throw new IllegalStateException("LiveListenerBus is stopped.")
    }

    queues.asScala.find(_.name == queue) match {
      case Some(queue) =>
        queue.addListener(listener)

      case None =>
        // it will create multiple AsyncEventQueues with the same name when run in multi-thread scene and those created AsyncEventQueues will be added to queues
        val newQueue = new AsyncEventQueue(queue, conf, metrics, this)
        newQueue.addListener(listener)
        if (started.get()) {
          newQueue.start(sparkContext)
        }
        queues.add(newQueue)
    }
  }

Issue Links

• links to: GitHub Pull Request #24676