[Hot Fix #469] Fix flaky test in SparkListenerSuite

[andrewor14]

The two modified tests may fail if the race condition does not bid in our favor...

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished. All automated tests passed.

[AmplabJenkins]

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14409/

[rxin]

Thanks. I've merged this.

[zsxwing]

Just realize that waitUntilEmpty is not enough. In waitUntilEmpty, it only checks eventQueue.isEmpty. But when eventQueue.isEmpty is true, there is still a chance that listeners do not finish their jobs or the memory is not synchronized.

[rxin]

cc @andrewor14

[andrewor14]

Good catch. A solution is to have eventQueue peek instead of take, such that we remove the event from the queue only after all listeners have finished processing it.

Though many other places have long been relying on waitUntilEmpty before this PR. It makes me wonder how likely this race condition actually causes a test failure... (nevertheless it's possible)

[andrewor14]

Hm, looks like LinkedBlockingQueue's take() is special in that it waits for the next item to be ready. It seems there isn't an equivalent for peek... we may have to synchronize some other way.

[zsxwing]

Since waitUntilEmpty is only for test, is it possible that moving such wait logic to the SparkListener instances used in tests? E.g.,

  import java.util.concurrent.{CountDownLatch, TimeUnit}

  class SaveStageAndTaskInfo extends SparkListener {
    val stageInfos = mutable.Map[StageInfo, Seq[(TaskInfo, TaskMetrics)]]()
    var taskInfoMetrics = mutable.Buffer[(TaskInfo, TaskMetrics)]()
    val latch = new CountDownLatch(1)

    override def onTaskEnd(task: SparkListenerTaskEnd) {
      val info = task.taskInfo
      val metrics = task.taskMetrics
      if (info != null && metrics != null) {
        taskInfoMetrics += ((info, metrics))
      }
    }

    override def onStageCompleted(stage: SparkListenerStageCompleted) {
      stageInfos(stage.stageInfo) = taskInfoMetrics
      taskInfoMetrics = mutable.Buffer.empty
      latch.countDown()
    }

    def waitForCompleted(timeoutMillis: Long) {
      latch.await(timeoutMillis, TimeUnit.MILLISECONDS)
    }
  }

[andrewor14]

@zsxwing we have to keep waitUntilEmpty in LiveListenerBus.scala (unfortunately) because DAGSchedulerSuite also uses it. I started a more general solution at #544. Maybe we can move the discussion there.