From d34bac0e156432ca6a260db73dbe1318060e309c Mon Sep 17 00:00:00 2001
From: zsxwing <zsxwing@gmail.com>
Date: Tue, 4 Aug 2015 20:09:15 -0700
Subject: [PATCH] [SPARK-9504] [STREAMING] [TESTS] Fix
o.a.s.streaming.StreamingContextSuite.stop gracefully again
The test failure is here: https://amplab.cs.berkeley.edu/jenkins/job/Spark-Master-SBT/3150/AMPLAB_JENKINS_BUILD_PROFILE=hadoop1.0,label=centos/testReport/junit/org.apache.spark.streaming/StreamingContextSuite/stop_gracefully/
There is a race condition in TestReceiver that it may add 1 record and increase `TestReceiver.counter` after stopping `BlockGenerator`. This PR just adds `join` to wait the pushing thread.
Author: zsxwing <zsxwing@gmail.com>
Closes #7934 from zsxwing/SPARK-9504-2 and squashes the following commits:
cfd7973 [zsxwing] Wait for the thread to make sure we won't change TestReceiver.counter after stopping BlockGenerator
---
.../org/apache/spark/streaming/StreamingContextSuite.scala | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala
index b7db280f63..7423ef6bcb 100644
--- a/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala
+++ b/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala
@@ -789,7 +789,8 @@ class TestReceiver extends Receiver[Int](StorageLevel.MEMORY_ONLY) with Logging
}
def onStop() {
- // no clean to be done, the receiving thread should stop on it own
+ // no clean to be done, the receiving thread should stop on it own, so just wait for it.
+ receivingThreadOption.foreach(_.join())
}
}
--
GitLab