dongjoon-hyun opened a new pull request, #60164: URL: https://github.com/apache/spark/pull/60164
### What changes were proposed in this pull request? `TaskContextImpl.invokePostStatusUpdateListeners()` no longer marks the task failed when a post-status-update listener throws. The exceptions are collected, all listeners still run, and a `TaskCompletionListenerException` is thrown afterwards so that `Executor` keeps logging it through `Utils.tryLogNonFatalError`, as before. The scaladoc of `TaskContext.addPostStatusUpdateListener` now says that listener failures are only logged. ### Why are the changes needed? `Executor.TaskRunner.run` invokes the post-status-update listeners (added by SPARK-57491 for the push-based shuffle block push) after it has sent `TaskState.FINISHED` to the driver. At that point `Task.run` has unset the `TaskContext`, and the executor has released the task's block locks and memory. `invokePostStatusUpdateListeners()` reuses `invokeListeners()`, which calls `markTaskFailed` for any listener that throws. So a failing post-status-update listener makes a task that was already reported as successful look failed: - `isFailed()` turns true and `getTaskFailure` is defined. - The task failure listeners run on the task thread, without a `TaskContext` and after the task's memory is released. In the repro, `TaskContext.get()` is `null` and the task memory is 0 inside the failure listener. The driver still has the successful result, so this is only visible to code that reads the task state or registers failure listeners (for example, `RocksDBStateStoreProvider` aborts a state store on failure, and a read-only store that was already released logs a misleading `STATE_STORE_OPERATION_OUT_OF_ORDER` error). The only in-tree post-status-update listener is the push-based shuffle one, so this is unlikely to be hit today. It is still wrong for the task to change state after the status update was sent. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added a test to `TaskContextSuite`. It fails without the change (a failure listener is invoked and `isFailed()` is true) and passes with it. ``` build/sbt 'core/testOnly *TaskContextSuite *ExecutorSuite *ShuffleBlockPusherSuite' ``` ### 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]
