dongjoon-hyun opened a new pull request, #59710: URL: https://github.com/apache/spark/pull/59710
### What changes were proposed in this pull request? This PR fixes the check that `AsyncProgressTrackingMicroBatchExecution.cleanup()` uses to decide whether the async log write executor stopped in time. SPARK-58131 added this check. - Check `isTerminated` instead of `isShutdown`. - If the executor has not terminated after the 30-second wait, log a warning and continue the cleanup instead of throwing `TimeoutException`. ### Why are the changes needed? `cleanup()` stops the executor and then checks `asyncWritesExecutorService.isShutdown`. `isShutdown` is true as soon as `shutdown()` or `shutdownNow()` is called, so the `else` branch can never run. - The warning and the `TimeoutException` are dead code. - The INFO message `... has been shutdown` is logged even when an offset or commit log write is still running after the 30-second wait. The write can keep running after the wait because `ThreadUtils.shutdown` does not call `shutdownNow()` after the grace period. SPARK-60466 fixes that separately. This PR still calls `ThreadUtils.shutdown`, so it gets that fix too. Making the `TimeoutException` reachable would be worse. `cleanup()` runs in the `finally` block of `StreamExecution.runStream`, before the rest of the query termination, so an exception from `cleanup()` skips that work. I tried it with a commit log write blocked during `query.stop()`: - `query.stop()` still returned normally after 30 seconds, because the exception is thrown in the stream execution thread and goes to its uncaught exception handler. - The query stayed in `spark.streams.active`, and no `QueryTerminatedEvent` was posted. - Restarting the query from the same checkpoint failed with `[CONCURRENT_QUERY] Another instance of this query was just started by a concurrent session.` ### Does this PR introduce _any_ user-facing change? Only a log message changes. When async log writes are still running 30 seconds after a query stops, Spark now logs the WARN message `... did not terminate in 30 seconds` instead of the INFO message `... has been shutdown`. Query stop behaves as before. SPARK-58131 is not in any released version yet. It is in `master`, `branch-4.x` and `branch-4.3`, and in the [v4.3.0-rc1](https://github.com/apache/spark/releases/tag/v4.3.0-rc1) (2026-08-31) and [v4.3.0-rc2](https://github.com/apache/spark/releases/tag/v4.3.0-rc2) (2026-10-12) tags. ### How was this patch tested? Pass the CIs with the newly added test case. The test blocks the async commit log write of a batch, stops the query, and then checks the log message and that the query is no longer active. Without this fix, the test fails because only the INFO message is logged. The test takes about 30 seconds because the wait in `cleanup()` is fixed at 30 seconds. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 5.5 -- 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]
