From dafdbb6c06dd2be49ede3fd6a4c745e0001bf272 Mon Sep 17 00:00:00 2001 From: Tathagata Das Date: Tue, 16 Aug 2016 16:40:28 -0700 Subject: [PATCH 1/2] Removed stacktrace from QueryTerminated --- .../sql/execution/streaming/StreamExecution.scala | 5 +---- .../sql/streaming/StreamingQueryException.scala | 3 ++- .../sql/streaming/StreamingQueryListener.scala | 3 +-- .../streaming/StreamingQueryListenerSuite.scala | 14 +++++++------- 4 files changed, 11 insertions(+), 14 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala index 66fb5a4bdeb7f..37262d7c21456 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala @@ -217,10 +217,7 @@ class StreamExecution( } finally { state = TERMINATED sparkSession.streams.notifyQueryTermination(StreamExecution.this) - postEvent(new QueryTerminated( - this.toInfo, - exception.map(_.getMessage), - exception.map(_.getStackTrace.toSeq).getOrElse(Nil))) + postEvent(new QueryTerminated(this.toInfo, exception.map(_.getMessage))) terminationLatch.countDown() } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryException.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryException.scala index 90f95ca9d4229..bd3e5a5618ec4 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryException.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryException.scala @@ -22,7 +22,8 @@ import org.apache.spark.sql.execution.streaming.{Offset, StreamExecution} /** * :: Experimental :: - * Exception that stopped a [[StreamingQuery]]. + * Exception that stopped a [[StreamingQuery]]. Use `cause` get the actual exception + * that caused the failure. * @param query Query that caused the exception * @param message Message of this exception * @param cause Internal cause of this exception diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala index 3b3cead3a66de..db606abb8ce43 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala @@ -108,6 +108,5 @@ object StreamingQueryListener { @Experimental class QueryTerminated private[sql]( val queryInfo: StreamingQueryInfo, - val exception: Option[String], - val stackTrace: Seq[StackTraceElement]) extends Event + val exception: Option[String]) extends 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 7f4d28cf0598f..9a1920dc5bcde 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 @@ -94,7 +94,6 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter { assert(status.id === query.id) assert(status.sourceStatuses(0).offsetDesc === Some(LongOffset(0).toString)) assert(status.sinkStatus.offsetDesc === CompositeOffset.fill(LongOffset(0)).toString) - assert(listener.terminationStackTrace.isEmpty) assert(listener.terminationException === None) } listener.checkAsyncErrors() @@ -147,7 +146,7 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter { } } - test("exception should be reported in QueryTerminated") { + testQuietly("exception should be reported in QueryTerminated") { val listener = new QueryStatusCollector withListenerAdded(listener) { val input = MemoryStream[Int] @@ -159,8 +158,12 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter { spark.sparkContext.listenerBus.waitUntilEmpty(10000) assert(listener.terminationStatus !== null) assert(listener.terminationException.isDefined) + // Make sure that the exception message reported through listener + // contains the actual exception and relevant stack trace + assert(!listener.terminationException.get.contains("StreamingQueryException")) + assert(listener.terminationException.get.contains("terminated with exception")) assert(listener.terminationException.get.contains("java.lang.ArithmeticException")) - assert(listener.terminationStackTrace.nonEmpty) + assert(listener.terminationException.get.contains("StreamingQueryListenerSuite")) } ) } @@ -205,8 +208,7 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter { val exception = new RuntimeException("exception") val queryQueryTerminated = new StreamingQueryListener.QueryTerminated( queryTerminatedInfo, - Some(exception.getMessage), - exception.getStackTrace) + Some(exception.getMessage)) val json = JsonProtocol.sparkEventToJson(queryQueryTerminated) val newQueryTerminated = JsonProtocol.sparkEventFromJson(json) @@ -262,7 +264,6 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter { @volatile var startStatus: StreamingQueryInfo = null @volatile var terminationStatus: StreamingQueryInfo = null @volatile var terminationException: Option[String] = null - @volatile var terminationStackTrace: Seq[StackTraceElement] = null val progressStatuses = new ConcurrentLinkedQueue[StreamingQueryInfo] @@ -296,7 +297,6 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter { assert(startStatus != null, "onQueryTerminated called before onQueryStarted") terminationStatus = queryTerminated.queryInfo terminationException = queryTerminated.exception - terminationStackTrace = queryTerminated.stackTrace } asyncTestWaiter.dismiss() } From 60eabcc4716f26e0c6d688d14b518f7321313f88 Mon Sep 17 00:00:00 2001 From: Tathagata Das Date: Tue, 16 Aug 2016 17:05:30 -0700 Subject: [PATCH 2/2] Convert both exception message and stack trace to string --- .../apache/spark/sql/execution/streaming/StreamExecution.scala | 2 +- .../spark/sql/streaming/StreamingQueryListenerSuite.scala | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala index 37262d7c21456..4d05af0b60358 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala @@ -217,7 +217,7 @@ class StreamExecution( } finally { state = TERMINATED sparkSession.streams.notifyQueryTermination(StreamExecution.this) - postEvent(new QueryTerminated(this.toInfo, exception.map(_.getMessage))) + postEvent(new QueryTerminated(this.toInfo, exception.map(_.cause).map(Utils.exceptionString))) terminationLatch.countDown() } } 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 9a1920dc5bcde..77602e8167fa3 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 @@ -161,7 +161,6 @@ class StreamingQueryListenerSuite extends StreamTest with BeforeAndAfter { // Make sure that the exception message reported through listener // contains the actual exception and relevant stack trace assert(!listener.terminationException.get.contains("StreamingQueryException")) - assert(listener.terminationException.get.contains("terminated with exception")) assert(listener.terminationException.get.contains("java.lang.ArithmeticException")) assert(listener.terminationException.get.contains("StreamingQueryListenerSuite")) }