Fix flaky AccumulatorSuite#verifyPeakExecutionMemorySet

Events are posted asynchronously so we may not have called
onStageCompleted or onJobEnd etc. before asserting things.
We should wait until all events have been processed before
proceeding.

Author: Andrew Or

core/src/test/scala/org/apache/spark/AccumulatorSuite.scala

@@ -307,6 +307,8 @@ private[spark] object AccumulatorSuite {
     val listener = new SaveInfoListener
     sc.addSparkListener(listener)
     testBody
+    // wait until all events have been processed before proceeding to assert things
+    sc.listenerBus.waitUntilEmpty(10 * 1000)
     val accums = listener.getCompletedStageInfos.flatMap(_.accumulables.values)
     val isSet = accums.exists { a =>
       a.name == Some(PEAK_EXECUTION_MEMORY) && a.value.exists(_.asInstanceOf[Long] > 0L)