From ad75ec0eb336f265d638ce97567984a464bae41d Mon Sep 17 00:00:00 2001 From: Eyal Zituny Date: Sun, 19 Feb 2017 12:55:30 +0200 Subject: [PATCH 1/3] [SPARK-19594] StreamingQueryListener fails to handle QueryTerminatedEvent if more then one listeners exists --- .../streaming/StreamingQueryListenerBus.scala | 5 +- .../StreamingQueryListenerSuite.scala | 87 +++++++++++++++++++ 2 files changed, 91 insertions(+), 1 deletion(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala index a2153d27e9fef..eba70731e148f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala +++ b/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 _ => } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenerSuite.scala index 4596aa1d348e3..f61d48fe0604c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenerSuite.scala +++ b/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() From 903f828732addc460e205adcb0d2ea22438d9371 Mon Sep 17 00:00:00 2001 From: Eyal Zituny Date: Wed, 22 Feb 2017 10:16:00 +0200 Subject: [PATCH 2/3] [SPARK-19594] StreamingQueryListener fails to handle QueryTerminatedEvent - CR fix --- .../org/apache/spark/util/ListenerBus.scala | 2 +- .../streaming/StreamingQueryListenerBus.scala | 16 +++++++++++++--- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/util/ListenerBus.scala b/core/src/main/scala/org/apache/spark/util/ListenerBus.scala index 79fc2e94599c7..fa5ad4e8d81e1 100644 --- a/core/src/main/scala/org/apache/spark/util/ListenerBus.scala +++ b/core/src/main/scala/org/apache/spark/util/ListenerBus.scala @@ -52,7 +52,7 @@ private[spark] trait ListenerBus[L <: AnyRef, E] extends Logging { * Post the event to all registered listeners. The `postToAll` caller should guarantee calling * `postToAll` in the same thread for all events. */ - final def postToAll(event: E): Unit = { + def postToAll(event: E): Unit = { // JavaConverters can create a JIterableWrapper if we use asScala. // However, this method will be called frequently. To avoid the wrapper cost, here we use // Java Iterator directly. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala index eba70731e148f..270c5a179b735 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala @@ -70,12 +70,22 @@ class StreamingQueryListenerBus(sparkListenerBus: LiveListenerBus) sparkListenerBus.post(s) // post to local listeners to trigger callbacks postToAll(s) + case _ => + sparkListenerBus.post(event) + } + } + + /** + * Post the event to all registered listeners. The `postToAll` caller should guarantee calling + * `postToAll` in the same thread for all events. also remove the query id after all listeners + * process the QueryTerminatedEvent + */ + override def postToAll(event: Event): Unit = { + super.postToAll(event) + event match { 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) } } From 896f5b33d01fa51db9d8ee53a8cc95274654f9e8 Mon Sep 17 00:00:00 2001 From: Eyal Zituny Date: Sun, 26 Feb 2017 09:53:48 +0200 Subject: [PATCH 3/3] [SPARK-19594] StreamingQueryListener fails to handle QueryTerminatedEvent - test and comments fix --- .../streaming/StreamingQueryListenerBus.scala | 5 +- .../StreamingQueryListenerSuite.scala | 72 ++----------------- 2 files changed, 7 insertions(+), 70 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala index 270c5a179b735..4207013c3f75d 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingQueryListenerBus.scala @@ -76,9 +76,8 @@ class StreamingQueryListenerBus(sparkListenerBus: LiveListenerBus) } /** - * Post the event to all registered listeners. The `postToAll` caller should guarantee calling - * `postToAll` in the same thread for all events. also remove the query id after all listeners - * process the QueryTerminatedEvent + * Override the parent `postToAll` to remove the query id from `activeQueryRunIds` after all + * the listeners process `QueryTerminatedEvent`. (SPARK-19594) */ override def postToAll(event: Event): Unit = { super.postToAll(event) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenerSuite.scala index f61d48fe0604c..eb09b9ffcfc5d 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenerSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenerSuite.scala @@ -133,50 +133,13 @@ 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) + test("SPARK-19594: all of listeners should receive QueryTerminatedEvent") { + val df = MemoryStream[Int].toDS().as[Long] + val listeners = (1 to 5).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 + StartStream(), StopStream, AssertOnQuery { query => eventually(Timeout(streamingTimeout)) { @@ -188,35 +151,10 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter { 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)) + listeners.foreach(spark.streams.removeListener) } }