[SPARK-19594][Structured Streaming] StreamingQueryListener fails to handle QueryTerminatedEvent if more then one listeners exists

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala

@@ -70,6 +70,10 @@ class StreamingQueryListenerBus(sparkListenerBus: LiveListenerBus)
         sparkListenerBus.post(s)
         // post to local listeners to trigger callbacks
         postToAll(s)
+      case t: QueryTerminatedEvent =>
+        // run all the listeners synchronized before removing the id from the list
+        postToAll(t)
+        activeQueryRunIds.synchronized { activeQueryRunIds -= t.runId }
       case _ =>
         sparkListenerBus.post(event)
     }
@@ -112,7 +116,6 @@ class StreamingQueryListenerBus(sparkListenerBus: LiveListenerBus)
       case queryTerminated: QueryTerminatedEvent =>
         if (shouldReport(queryTerminated.runId)) {
           listener.onQueryTerminated(queryTerminated)
-          activeQueryRunIds.synchronized { activeQueryRunIds -= queryTerminated.runId }
         }
       case _ =>
     }

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenerSuite.scala

@@ -133,6 +133,93 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter {
     }
   }
 
+  testQuietly("multiple listeners, check trigger events are generated correctly") {
+    val clock = new StreamManualClock
+    val inputData = new MemoryStream[Int](0, sqlContext)
+    val df = inputData.toDS().as[Long].map { 10 / _ }
+    val numOfListenersToTest = 5
+    val listeners = (1 to numOfListenersToTest).map(_ => new EventCollector)
+    try {
+      // No events until started
+      listeners.foreach(listener => spark.streams.addListener(listener))
+      listeners.foreach(listener => assert(listener.startEvent === null))
+      listeners.foreach(listener => assert(listener.progressEvents.isEmpty))
+        listeners.foreach(listener => assert(listener.terminationEvent === null))
+
+      testStream(df, OutputMode.Append)(
+
+        // Start event generated when query started
+        StartStream(ProcessingTime(100), triggerClock = clock),
+        AssertOnQuery { query =>
+          listeners.foreach(listener => assert(listener.startEvent !== null))
+          listeners.foreach(listener => assert(listener.startEvent.id === query.id))
+          listeners.foreach(listener => assert(listener.startEvent.runId === query.runId))
+          listeners.foreach(listener => assert(listener.startEvent.name === query.name))
+          listeners.foreach(listener => assert(listener.progressEvents.isEmpty))
+          listeners.foreach(listener => assert(listener.terminationEvent === null))
+          true
+        },
+
+        // Progress event generated when data processed
+        AddData(inputData, 1, 2),
+        AdvanceManualClock(100),
+        CheckAnswer(10, 5),
+        AssertOnQuery { query =>
+          listeners.foreach(listener => assert(listener.progressEvents.nonEmpty))
+          // SPARK-18868: We can't use query.lastProgress, because in progressEvents, we filter
+          // out non-zero input rows, but the lastProgress may be a zero input row trigger
+          val lastNonZeroProgress = query.recentProgress.filter(_.numInputRows > 0).lastOption
+            .getOrElse(fail("No progress updates received in StreamingQuery!"))
+          listeners.foreach(listener =>
+            assert(listener.progressEvents.last.json === lastNonZeroProgress.json))
+          listeners.foreach(listener => assert(listener.terminationEvent === null))
+          true
+        },
+
+        // Termination event generated when stopped cleanly
+        StopStream,
+        AssertOnQuery { query =>
+          eventually(Timeout(streamingTimeout)) {
+            listeners.foreach(listener => assert(listener.terminationEvent !== null))
+            listeners.foreach(listener => assert(listener.terminationEvent.id === query.id))
+            listeners.foreach(listener => assert(listener.terminationEvent.runId === query.runId))
+            listeners.foreach(listener => assert(listener.terminationEvent.exception === None))
+          }
+          listeners.foreach(listener => listener.checkAsyncErrors())
+          listeners.foreach(listener => listener.reset())
+          true
+        },
+
+        // Termination event generated with exception message when stopped with error
+        StartStream(ProcessingTime(100), triggerClock = clock),
+        AddData(inputData, 0),
+        AdvanceManualClock(100),
+        ExpectFailure[SparkException](),
+        AssertOnQuery { query =>
+          eventually(Timeout(streamingTimeout)) {
+            listeners.foreach(listener => assert(listener.terminationEvent !== null))
+            listeners.foreach(listener => assert(listener.terminationEvent.id === query.id))
+            listeners.foreach(listener => assert(listener.terminationEvent.exception.nonEmpty))
+            // Make sure that the exception message reported through listener
+            // contains the actual exception and relevant stack trace
+            listeners.foreach(listener =>
+              assert(!listener.terminationEvent.exception.get.contains("StreamingQueryException")))
+            listeners.foreach(listener =>
+              assert(
+                listener.terminationEvent.exception.get.contains("java.lang.ArithmeticException")))
+            listeners.foreach(listener =>
+              assert(
+                listener.terminationEvent.exception.get.contains("StreamingQueryListenerSuite")))
+          }
+          listeners.foreach(listener => listener.checkAsyncErrors())
+          true
+        }
+      )
+    } finally {
+      listeners.foreach(listener => spark.streams.removeListener(listener))
+    }
+  }
+
   test("adding and removing listener") {
     def isListenerActive(listener: EventCollector): Boolean = {
       listener.reset()