snmvaughan opened a new pull request, #59645: URL: https://github.com/apache/spark/pull/59645
### What changes were proposed in this pull request? Backport of #55676 to branch-4.0, adapted to this branch: - branch-4.0 has no `AsyncStreamingQueryCheckpointMetadata`. `AsyncProgressTrackingMicroBatchExecution` creates `AsyncOffsetSeqLog` and `AsyncCommitLog` itself, so it passes `errorNotifier` to them directly. `errorNotifier` is a `val` of `MicroBatchExecution`, so it is initialized before these logs are. - The streaming classes are still in the `org.apache.spark.sql.execution.streaming` package on this branch. So the `runtime.ErrorNotifier` imports in the two logs, and the new `checkpointing`/`runtime` imports in the suite, are not needed and are left out. Otherwise the change is identical to #55676. `AsyncOffsetSeqLog` and `AsyncCommitLog` now share the query's `ErrorNotifier`. After a write has failed: - Offset and commit log write tasks that run later fail with the first error and don't write to the checkpoint. - `ErrorNotifier.markError` keeps the first error and attaches later ones to it as suppressed exceptions, so a cascading failure can't replace the root cause. ### Why are the changes needed? With async progress tracking, offset and commit log entries are written by a single-threaded executor. When the write of `offsets/N` fails, a `commits/N` write queued after it can still succeed. That leaves a commit log entry without its offset log entry in the checkpoint. The query can also fail with a later, cascading error instead of the I/O error that started it. The same gap exists on branch-4.0. SPARK-60264 (#59518), which is already on branch-4.0, makes `AbortableStreamBasedCheckpointFileManager` report failed writes instead of only logging them. With async progress tracking, a commit log write queued behind such a failure can still go through without this change. ### Does this PR introduce _any_ user-facing change? No. As in #55676, a query that hits a write failure still stops with a `StreamingQueryException`, which now reports the original error. ### How was this patch tested? Includes the regression test from #55676, "async log writes record first error and gate subsequent writes", and its assertion change in `testAsyncWriteErrorsPermissionsIssue`, both in `AsyncProgressTrackingMicroBatchExecutionSuite`. The changed main and test sources compile against the Spark 4.0.4 release jars with this branch's scalac options, and pass scalastyle. The suite was not run locally; this PR's CI runs it. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (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]
