From e6ceac49a311faf3413acda57a6612fe806adf90 Mon Sep 17 00:00:00 2001
From: Andrew Or <andrew@databricks.com>
Date: Fri, 29 Jan 2016 17:59:41 -0800
Subject: [PATCH] [SPARK-13096][TEST] 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.
https://amplab.cs.berkeley.edu/jenkins/job/spark-master-test-sbt-hadoop-2.7/79/testReport/junit/org.apache.spark.util.collection/ExternalAppendOnlyMapSuite/spilling/
Author: Andrew Or <andrew@databricks.com>
Closes #10990 from andrewor14/accum-suite-less-flaky.
---
core/src/test/scala/org/apache/spark/AccumulatorSuite.scala | 2 ++
1 file changed, 2 insertions(+)
diff --git a/core/src/test/scala/org/apache/spark/AccumulatorSuite.scala b/core/src/test/scala/org/apache/spark/AccumulatorSuite.scala
index 11c97d7d9a..b8f2b96d70 100644
--- a/core/src/test/scala/org/apache/spark/AccumulatorSuite.scala
+++ b/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)
--
GitLab