aboutsummaryrefslogtreecommitdiff
path: root/core
diff options
context:
space:
mode:
authorAndrew Or <andrewor14@gmail.com>2014-04-23 21:59:33 -0700
committerReynold Xin <rxin@apache.org>2014-04-23 21:59:33 -0700
commit4b2bab1d08a6b790be94717bbdd643d896d85c16 (patch)
tree89447fabc5248811d7d23b4fb79ae83aaa586ed3 /core
parent640f9a0efefd42cff86aecd4878a3a57f5ae85fa (diff)
downloadspark-4b2bab1d08a6b790be94717bbdd643d896d85c16.tar.gz
spark-4b2bab1d08a6b790be94717bbdd643d896d85c16.tar.bz2
spark-4b2bab1d08a6b790be94717bbdd643d896d85c16.zip
[Hot Fix #469] Fix flaky test in SparkListenerSuite
The two modified tests may fail if the race condition does not bid in our favor... Author: Andrew Or <andrewor14@gmail.com> Closes #516 from andrewor14/stage-info-test-fix and squashes the following commits: b4b6100 [Andrew Or] Add/replace missing waitUntilEmpty() calls to listener bus
Diffstat (limited to 'core')
-rw-r--r--core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala5
1 files changed, 3 insertions, 2 deletions
diff --git a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
index ab139175e0..ba048ced32 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
@@ -50,9 +50,9 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatc
(1 to 5).foreach { _ => bus.post(SparkListenerJobEnd(0, JobSucceeded)) }
assert(counter.count === 0)
- // Starting listener bus should flush all buffered events (asynchronously, hence the sleep)
+ // Starting listener bus should flush all buffered events
bus.start()
- Thread.sleep(1000)
+ assert(bus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS))
assert(counter.count === 5)
// After listener bus has stopped, posting events should not increment counter
@@ -177,6 +177,7 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatc
listener.stageInfos.clear()
rdd3.count()
+ assert(sc.listenerBus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS))
listener.stageInfos.size should be {2} // Shuffle map stage + result stage
val stageInfo3 = listener.stageInfos.keys.find(_.stageId == 2).get
stageInfo3.rddInfos.size should be {2} // ShuffledRDD, MapPartitionsRDD