eason-yuchen-liu opened a new pull request, #59727: URL: https://github.com/apache/spark/pull/59727
### What changes were proposed in this pull request? This PR makes a streaming query finish terminating when releasing its resources fails, and reports that failure as a query failure. - `StreamExecution.runStream` runs `stopSources()` and `cleanup()` each in their own `try`. If one fails, the failure becomes the query's failure (`STREAM_FAILED`). If the query had already failed, it is added to that failure as suppressed instead. The rest of the termination then still runs: the query is set to `TERMINATED`, unregistered from `StreamingQueryManager`, and `QueryTerminatedEvent` is posted with the failure. A fatal error is rethrown at the end, so it still reaches the `UncaughtExceptionHandler` like a fatal query failure. - The code that builds the detailed `StreamingQueryException` moves into `detailedStreamDeathCause`, so both paths build the same exception. - `AsyncProgressTrackingMicroBatchExecution.cleanup()` runs `super.cleanup()` even if committing the queued source offsets fails. The commit is still attempted, and its failure still fails the query. ### Why are the changes needed? `stopSources()` and `cleanup()` run in the `finally` block of `StreamExecution.runStream`, at the start of the query termination. An exception from either one skips everything after it: - The query is never set to `TERMINATED`, so `isActive` stays true. - The query is never unregistered. It stays in `spark.streams.active`, and restarting it from the same checkpoint fails with `CONCURRENT_QUERY` until the driver restarts. `spark.streams.awaitAnyTermination()` never returns. - No `QueryTerminatedEvent` is posted. - The failure is not reported. A query stopped with `stop()` looks stopped cleanly: `exception` is `None` and `awaitTermination()` returns normally. The error only reaches the stream thread's uncaught exception handler. SPARK-60469 saw the same consequences and, for that reason, logs a warning instead of throwing when the async log write executor does not terminate in time. This PR fixes the termination itself, so cleanup code no longer has to avoid throwing. Two paths can throw: - `AsyncProgressTrackingMicroBatchExecution.cleanup()` commits the source offsets that are still queued (added by SPARK-58131), so a source whose `commit` throws while the query stops hits this path. That also skips `super.cleanup()`, which leaks the async log purge executor. - `stopSources()` only catches non-fatal errors, so a fatal error from `Source.stop()` hits this path for any query. ### Does this PR introduce _any_ user-facing change? Yes. When releasing a streaming query's resources fails during termination, the query now terminates as failed: - `exception` and `awaitTermination()` report the failure. - `QueryTerminatedEvent` carries it. - The query is no longer active and can be restarted. Before, the query stayed registered as active, could not be restarted on the same driver, and the failure was not reported. The source commit path only exists in unreleased code (SPARK-58131 is in `master`, `branch-4.x` and `branch-4.3`). A fatal error from `Source.stop()` reaches this path in released versions too. ### How was this patch tested? New tests: - `AsyncProgressTrackingMicroBatchExecutionSuite`: "a source commit failure on stop fails the query and still terminates it". The test blocks the offset log write that queues a source commit until the next batch has started, so the commit is left to `cleanup()`. Then it makes the source commit throw, stops the query, and checks three things: the query reports the failure, it is no longer active, and restarting it from the same checkpoint works. - `StreamSuite`: "a fatal error while stopping the sources fails the query and still terminates it". A source's `stop()` throws a `StackOverflowError` when the query is stopped. The test checks that the query reports the error, is unregistered, and posts `QueryTerminatedEvent`. It also checks that the error reaches the uncaught exception handler. - `StreamSuite`: "an error while stopping the sources of a failed query is suppressed". The query fails in `getBatch`, then its source's `stop()` throws. The test checks that the original failure is still reported, with the stop error suppressed, and that the query still terminates. All three tests fail without the fix and pass with it. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 5.5 This pull request and its description were written by Isaac. -- 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]
