SPARK-2113: awaitTermination() after stop() will hang in Spark Stremaing

Author: Lars Albertsson <lalle@spotify.com> Closes #1001 from lallea/contextwaiter_stopped and squashes the following commits: 93cd314 [Lars Albertsson] Mend StreamingContext stop() followed by awaitTermination().
apache · Jun 11, 2014 · 4d5c12a · 4d5c12a
1 parent e508f59
commit 4d5c12a
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 0 deletions.
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ContextWaiter.scala b/streaming/src/main/scala/org/apache/spark/streaming/ContextWaiter.scala
@@ -27,6 +27,7 @@ private[streaming] class ContextWaiter {
   }
 
   def notifyStop() = synchronized {
+    stopped = true
     notifyAll()
   }
 

diff --git a/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala
@@ -223,6 +223,18 @@ class StreamingContextSuite extends FunSuite with BeforeAndAfter with Timeouts w
     }
   }
 
+  test("awaitTermination after stop") {
+    ssc = new StreamingContext(master, appName, batchDuration)
+    val inputStream = addInputStream(ssc)
+    inputStream.map(x => x).register()
+
+    failAfter(10000 millis) {
+      ssc.start()
+      ssc.stop()
+      ssc.awaitTermination()
+    }
+  }
+
   test("awaitTermination with error in task") {
     ssc = new StreamingContext(master, appName, batchDuration)
     val inputStream = addInputStream(ssc)