HyukjinKwon commented on code in PR #59276:
URL: https://github.com/apache/spark/pull/59276#discussion_r4227593002
##########
sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala:
##########
@@ -446,6 +447,62 @@ class StreamSuite extends StreamTest {
}
}
+ /** Also returns the error, if any, rethrown to the query thread's
`UncaughtExceptionHandler`. */
+ private def failWhileBuildingQueryException(
+ buildFailure: Throwable): (StreamingQueryException, Option[Throwable]) =
{
+ val source = new Source {
+ @volatile private var failed = false
+
+ override def getOffset: Option[Offset] = Some(LongOffset(0))
+
+ override def getBatch(start: Option[Offset], end: Offset): DataFrame = {
+ failed = true
+ throw new IllegalStateException("getBatch failed")
+ }
+
+ override def schema: StructType = StructType(Array(StructField("value",
IntegerType)))
+
+ override def stop(): Unit = {}
+
+ // The query debug string embeds the source, so this fails building the
detailed exception.
+ override def toString: String = if (failed) throw buildFailure else
"FailingSource"
+ }
+ val df = Dataset[Int](
+ sqlContext.sparkSession,
+ StreamingExecutionRelation(source, sqlContext.sparkSession))
+ val query = df.writeStream.format("noop").start()
+ val queryThread =
query.asInstanceOf[StreamingQueryWrapper].streamingQuery.queryExecutionThread
+ val uncaught = new AtomicReference[Throwable]()
+ queryThread.setUncaughtExceptionHandler((_: Thread, e: Throwable) =>
uncaught.set(e))
Review Comment:
**Nit (P3):** `start()` only waits for `startLatch`, which `runStream`
releases right after posting `QueryStartedEvent` and before the first
micro-batch runs. After that, the stream thread runs on its own: it calls
`getBatch`, fails to build the detailed exception, and rethrows the
`StackOverflowError`. If this thread is descheduled between `start()` returning
and this call (for example on a loaded CI runner), the error reaches the
default handler instead, `uncaught` stays null, and
`assert(uncaught.contains(buildFailure))` fails even though the production code
behaved correctly. In the same window, the non-fatal test's `uncaught.isEmpty`
passes trivially.
`StreamTest.testStream` avoids this by installing the handler from
`StreamingQueryListener.onQueryStarted`, which runs synchronously on the stream
thread before `startLatch` is released (see the comment in `StreamTest.scala`
around line 418). Doing the same here, by adding the listener before `start()`
and removing it in `finally`, or having `getBatch` block on a latch that the
test releases after installing the handler, would make the test deterministic.
**Verification:**
- **Regression:** Both SPARK-60091 tests pass, and the fatal case records
the StackOverflowError through the handler even if the test thread is delayed
after start() returns.
- **Inspection:** The handler is installed on the stream thread from
onQueryStarted, which is ordered before startLatch is released, and no handler
installation remains after start().
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
