Wait for all event notifications before asserting results

robbins · robbins · commit f1df02b6834b · 2015-09-03T18:15:56.000+01:00
diff --git a/streaming/src/test/scala/org/apache/spark/streaming/InputStreamsSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/InputStreamsSuite.scala
@@ -76,6 +76,9 @@ class InputStreamsSuite extends TestSuiteBase with BeforeAndAfter {
           fail("Timeout: cannot finish all batches in 30 seconds")
         }
 
+        // Ensure progress listener has been notified of all events
+        ssc.scheduler.listenerBus.waitUntilEmpty(500)
+
         // Verify all "InputInfo"s have been reported
         assert(ssc.progressListener.numTotalReceivedRecords === input.size)
         assert(ssc.progressListener.numTotalProcessedRecords === input.size)

Original file line number	Diff line number	Diff line change
`@@ -76,6 +76,9 @@ class InputStreamsSuite extends TestSuiteBase with BeforeAndAfter {`
`76`	`76`	`fail("Timeout: cannot finish all batches in 30 seconds")`
`77`	`77`	`}`
`78`	`78`
	`79`	`+ // Ensure progress listener has been notified of all events`
	`80`	`+ ssc.scheduler.listenerBus.waitUntilEmpty(500)`
	`81`	`+`
`79`	`82`	`// Verify all "InputInfo"s have been reported`
`80`	`83`	`assert(ssc.progressListener.numTotalReceivedRecords === input.size)`
`81`	`84`	`assert(ssc.progressListener.numTotalProcessedRecords === input.size)`