[SPARK-18908][SS] Creating StreamingQueryException should check if logicalPlan is created #16322

zsxwing · 2016-12-17T01:20:00Z

What changes were proposed in this pull request?

This PR audits places using logicalPlan in StreamExecution and ensures they all handles the case that logicalPlan cannot be created.

In addition, this PR also fixes the following issues in StreamingQueryException:

StreamingQueryException and StreamExecution are cycle-dependent because in the StreamingQueryException's constructor, it calls StreamExecution's toDebugString which uses StreamingQueryException. Hence it will output null value in the error message.
Duplicated stack trace when calling Throwable.printStackTrace because StreamingQueryException's toString contains the stack trace.

How was this patch tested?

The updated test("max files per trigger - incorrect values"). I found this issue when I switched from testStream to the real codes to verify the failure in this test.

…er trigger - incorrect values

zsxwing · 2016-12-17T01:20:33Z

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

@@ -560,6 +589,9 @@ class StreamExecution(
  @volatile private var noNewData = false

  override def processAllAvailable(): Unit = {
+    if (streamDeathCause != null) {


Just make processAllAvailable fail fast.

SparkQA · 2016-12-17T03:26:52Z

Test build #70288 has finished for PR 16322 at commit 183be5a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-12-19T07:07:34Z

Test build #70342 has started for PR 16322 at commit 2044bc0.

zsxwing · 2016-12-19T20:18:31Z

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

@@ -240,7 +241,7 @@ trait StreamTest extends QueryTest with SharedSQLContext with Timeouts {
    val resetConfValues = mutable.Map[String, Option[String]]()

    @volatile
-    var streamDeathCause: Throwable = null
+    var streamThreadDeathCause: Throwable = null


Renamed it to avoid confusing with StreamExecution.streamDeathCause

zsxwing · 2016-12-19T20:21:15Z

cc @tdas @brkyvz @marmbrus

marmbrus · 2016-12-19T20:30:07Z

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

+      }
+    }
+    _logicalPlan
+  }


This is getting pretty complicated... Do we really need to include all of this information in StreamingQueryException?

all of this information

Do you mean logicalPlan? It's used to generate the plan for each batch.

What I'm trying to say is, (if I understand correctly) all of this extra logic is to make sure that there aren't cycles when we fail to initialize and then try to use that uninitialized information in order to generate the exception.

I'm asking why that information needs to be in the exception at all? Or if we should just try to initialize earlier in the stream thread and throw a different exception when initialization fails.

I guess it follows AnalysisException to include as many as information as possible. @tdas do you want to clarify it?

Sure, if we separate initialization failure from runtime failures though, then we can only include the extra information when it makes sense (i.e. when we have actually started running and have offsets, etc to report).

I agree with @marmbrus. We can include logicalPlan in the toDebugString only if the state allows, e.g. microBatchThreadStatus = INITIALIZED, and then we don't need any of this extra code

or really what i was suggesting was to have a separate try/catch early in the stream thread that forces initialization of anything that is lazy.

SparkQA · 2016-12-19T21:18:13Z

Test build #70372 has finished for PR 16322 at commit ccb2a51.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-12-19T22:29:35Z

Test build #70376 has finished for PR 16322 at commit 8fbb387.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-12-19T22:30:49Z

Test build #70377 has finished for PR 16322 at commit fb74f62.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-12-19T22:51:20Z

Test build #70375 has finished for PR 16322 at commit 185f177.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-12-20T00:43:45Z

Test build #70381 has finished for PR 16322 at commit 299a13d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

marmbrus · 2016-12-20T02:25:46Z

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

@@ -282,7 +289,7 @@ class StreamExecution(
        updateStatusMessage("Stopped")
      case e: Throwable =>
        streamDeathCause = new StreamingQueryException(
-          this,
+          toDebugString(includeLogicalPlan = logicalPlanInitilized),
          s"Query $prettyIdString terminated with exception: ${e.getMessage}",
          e,
          committedOffsets.toOffsetSeq(sources, offsetSeqMetadata).toString,


I think the one possibly confusing thing is that we are still wrapping initialization errors with this class that has a lot of field that don't make sense until the stream is initialized (i.e. committed and available offsets). I guess you did it this way because the function is Option[StreamingQueryException]? We might consider changing that, but others should chime in if they feel strongly.

Right. StreamingQueryException is a public API. I think it's better to not break it considering 2.1.0 just passed.

zsxwing · 2016-12-20T22:07:57Z

@marmbrus updated as per our offline discussion. I also added a new method assertAwaitThread to avoid the user calling the awaitXXX methods inside the stream thread (e.g., in the StreamingQueryListener.onQueryStarted)

SparkQA · 2016-12-21T01:21:33Z

Test build #70425 has finished for PR 16322 at commit 38dab48.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tdas · 2016-12-21T22:53:59Z

@marmbrus should take a look because he had opinions on this

marmbrus · 2016-12-21T23:02:27Z

LGTM

tdas · 2016-12-22T01:36:09Z

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

                }
              })
+            currentStream.awaitInitialization(streamingTimeout.toMillis)


Can you document why this is needed?

tdas · 2016-12-22T01:38:57Z

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

@@ -559,7 +580,36 @@ class StreamExecution(
  /** A flag to indicate that a batch has completed with no new data available. */
  @volatile private var noNewData = false

+  /**
+   * Assert that the await APIs should not be called in the stream thread. Otherwise, it may cause
+   * dead-lock.


Document what kind of deadlock can happen.

tdas · 2016-12-22T01:41:52Z

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

+   */
+  private def assertAwaitThread(): Unit = {
+    if (microBatchThread eq Thread.currentThread) {
+      throw new IllegalStateException("Cannot wait inside a stream thread")


cannot wait for a query state from the same thread that is running the query

tdas · 2016-12-22T01:52:13Z

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

@@ -228,7 +228,8 @@ trait StreamTest extends QueryTest with SharedSQLContext with Timeouts {
   */
  def testStream(
      _stream: Dataset[_],
-      outputMode: OutputMode = OutputMode.Append)(actions: StreamAction*): Unit = {
+      outputMode: OutputMode = OutputMode.Append,
+      ignoreThreadDeathCause: Boolean = false)(actions: StreamAction*): Unit = {


This ignore field is really hard to understand as a user of this function. And adding a field to just for one test does not make sense. Rather make an action ExpectStreamThreadException.

tdas

LGTM, pending tests

SparkQA · 2016-12-22T05:05:27Z

Test build #70502 has finished for PR 16322 at commit 8500e16.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2016-12-22T06:02:31Z

Thanks! Merging to master and 2.1.

…gicalPlan is created ## What changes were proposed in this pull request? This PR audits places using `logicalPlan` in StreamExecution and ensures they all handles the case that `logicalPlan` cannot be created. In addition, this PR also fixes the following issues in `StreamingQueryException`: - `StreamingQueryException` and `StreamExecution` are cycle-dependent because in the `StreamingQueryException`'s constructor, it calls `StreamExecution`'s `toDebugString` which uses `StreamingQueryException`. Hence it will output `null` value in the error message. - Duplicated stack trace when calling Throwable.printStackTrace because StreamingQueryException's toString contains the stack trace. ## How was this patch tested? The updated `test("max files per trigger - incorrect values")`. I found this issue when I switched from `testStream` to the real codes to verify the failure in this test. Author: Shixiong Zhu <shixiong@databricks.com> Closes #16322 from zsxwing/SPARK-18907. (cherry picked from commit ff7d82a) Signed-off-by: Shixiong Zhu <shixiong@databricks.com>

…gicalPlan is created ## What changes were proposed in this pull request? This PR audits places using `logicalPlan` in StreamExecution and ensures they all handles the case that `logicalPlan` cannot be created. In addition, this PR also fixes the following issues in `StreamingQueryException`: - `StreamingQueryException` and `StreamExecution` are cycle-dependent because in the `StreamingQueryException`'s constructor, it calls `StreamExecution`'s `toDebugString` which uses `StreamingQueryException`. Hence it will output `null` value in the error message. - Duplicated stack trace when calling Throwable.printStackTrace because StreamingQueryException's toString contains the stack trace. ## How was this patch tested? The updated `test("max files per trigger - incorrect values")`. I found this issue when I switched from `testStream` to the real codes to verify the failure in this test. Author: Shixiong Zhu <shixiong@databricks.com> Closes apache#16322 from zsxwing/SPARK-18907.

zsxwing added 2 commits December 16, 2016 16:16

Fix flaky test: o.a.s.sql.streaming.FileStreamSourceSuite max files p…

11a27e9

…er trigger - incorrect values

Fix the crash when StreamExecution fails to create the logical plan

183be5a

zsxwing changed the title ~~[SPARK-18908] Creating StreamingQueryException should check if logicalPlan is created~~ [SPARK-18908][SS] Creating StreamingQueryException should check if logicalPlan is created Dec 17, 2016

zsxwing commented Dec 17, 2016

View reviewed changes

Fix comment style

2044bc0

zsxwing added 3 commits December 19, 2016 10:54

Update tests

ccb2a51

Fix StreamingQueryException

185f177

Fix more

8fbb387

zsxwing commented Dec 19, 2016

View reviewed changes

Minor

fb74f62

marmbrus reviewed Dec 19, 2016

View reviewed changes

Address

299a13d

marmbrus reviewed Dec 20, 2016

View reviewed changes

zsxwing added 2 commits December 20, 2016 14:02

Update as per offline discussion

7d09656

Remove unnecessary changes

38dab48

tdas reviewed Dec 22, 2016

View reviewed changes

Address TD's comments

8500e16

tdas reviewed Dec 22, 2016

View reviewed changes

asfgit closed this in ff7d82a Dec 22, 2016

zsxwing deleted the SPARK-18907 branch December 22, 2016 06:06

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-18908][SS] Creating StreamingQueryException should check if logicalPlan is created #16322

[SPARK-18908][SS] Creating StreamingQueryException should check if logicalPlan is created #16322

zsxwing commented Dec 17, 2016 •

edited

zsxwing Dec 17, 2016

SparkQA commented Dec 17, 2016

SparkQA commented Dec 19, 2016

zsxwing Dec 19, 2016

zsxwing commented Dec 19, 2016

marmbrus Dec 19, 2016

zsxwing Dec 19, 2016

marmbrus Dec 19, 2016

zsxwing Dec 19, 2016

marmbrus Dec 19, 2016

brkyvz Dec 19, 2016

marmbrus Dec 19, 2016

SparkQA commented Dec 19, 2016

SparkQA commented Dec 19, 2016

SparkQA commented Dec 19, 2016

SparkQA commented Dec 19, 2016

SparkQA commented Dec 20, 2016

marmbrus Dec 20, 2016

zsxwing Dec 20, 2016

zsxwing commented Dec 20, 2016

SparkQA commented Dec 21, 2016

tdas commented Dec 21, 2016

marmbrus commented Dec 21, 2016

tdas Dec 22, 2016

tdas Dec 22, 2016

tdas Dec 22, 2016 •

edited

tdas Dec 22, 2016

tdas left a comment

SparkQA commented Dec 22, 2016

zsxwing commented Dec 22, 2016

[SPARK-18908][SS] Creating StreamingQueryException should check if logicalPlan is created #16322

[SPARK-18908][SS] Creating StreamingQueryException should check if logicalPlan is created #16322

Conversation

zsxwing commented Dec 17, 2016 • edited

What changes were proposed in this pull request?

How was this patch tested?

Choose a reason for hiding this comment

SparkQA commented Dec 17, 2016

SparkQA commented Dec 19, 2016

Choose a reason for hiding this comment

zsxwing commented Dec 19, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 19, 2016

SparkQA commented Dec 19, 2016

SparkQA commented Dec 19, 2016

SparkQA commented Dec 19, 2016

SparkQA commented Dec 20, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zsxwing commented Dec 20, 2016

SparkQA commented Dec 21, 2016

tdas commented Dec 21, 2016

marmbrus commented Dec 21, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tdas Dec 22, 2016 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tdas left a comment

Choose a reason for hiding this comment

SparkQA commented Dec 22, 2016

zsxwing commented Dec 22, 2016

zsxwing commented Dec 17, 2016 •

edited

tdas Dec 22, 2016 •

edited