Fix flaky verifyPeakExecutionMemorySet

Previously we would assert things before all events are guaranteed
to have been processed. To fix this, just block until all events
are actually processed, i.e. until the listener queue is empty.

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)