[SPARK-9869] [STREAMING] Wait for all event notifications before asserting results

robbins · Marcelo Vanzin · commit 460d3c5b0bdf · 2015-09-29T13:19:44.000-07:00
Author: robbins <robbins@uk.ibm.com> Closes apache#8589 from robbinspg/InputStreamSuite-fix. (cherry picked from commit 754f853) Signed-off-by: Andrew Or <andrew@databricks.com> (cherry picked from commit f945b64)
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)`