Update comments

apache · Jan 9, 2017 · fca0424 · fca0424
1 parent 59a1161
commit fca0424
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 4 deletions.
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
@@ -238,7 +238,7 @@ class StreamSuite extends StreamTest {
     }
   }
 
-  testQuietly("handle fatal errors thrown from the stream thread correctly") {
+  testQuietly("handle fatal errors thrown from the stream thread") {
     for (e <- Seq(
       new VirtualMachineError {},
       new ThreadDeath,

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamTest.scala
@@ -236,9 +236,9 @@ trait StreamTest extends QueryTest with SharedSQLContext with Timeouts {
   def testStream(
       _stream: Dataset[_],
       outputMode: OutputMode = OutputMode.Append)(actions: StreamAction*): Unit = synchronized {
-    // `synchronized` is added to prevent the user from calling `testStream` concurrently because
-    // this method uses `StreamingQueryListener` and it may not work correctly when `testStream`
-    // runs concurrently.
+    // `synchronized` is added to prevent the user from calling multiple `testStream`s concurrently
+    // because this method assumes there is only one active query in its `StreamingQueryListener`
+    // and it may not work correctly when multiple `testStream`s run concurrently.
 
     val stream = _stream.toDF()
     val sparkSession = stream.sparkSession  // use the session in DF, not the default session
@@ -255,6 +255,7 @@ trait StreamTest extends QueryTest with SharedSQLContext with Timeouts {
     // during query initialization.
     val listener = new StreamingQueryListener {
       override def onQueryStarted(event: QueryStartedEvent): Unit = {
+        // Note: this assumes there is only one query active in the `testStream` method.
         Thread.currentThread.setUncaughtExceptionHandler(new UncaughtExceptionHandler {
           override def uncaughtException(t: Thread, e: Throwable): Unit = {
             streamThreadDeathCause = e