driessamyn-sparta opened a new pull request, #29676:
URL: https://github.com/apache/flink/pull/29676
## What is the purpose of the change
Since FLINK-38180, the clean-up of a failed task runs in
`Task#postFailureCleanUpRegistry`, which `Task#doRun` closes inside the try
block that reports "exception in exception handler" through `notifyFatalError`.
An exception thrown by `AbstractInvokable#cleanUp`, for example from an
operator or sink writer `close()`, therefore shuts the TaskManager down and
takes every other task on it with it. Up to 2.1, `restoreAndInvoke` added such
an exception as suppressed to the original failure and the task simply failed.
This PR restores that behaviour while keeping the FLINK-38180 ordering. See
FLINK-41269 for the production trigger (`KafkaWriter#close()` rethrowing a
pending producer failure on Flink 2.2.0).
## Brief change log
- `Task#restoreAndInvoke` registers a new private method
`cleanUpAfterFailure(invokable, failure)` in the post-failure registry, instead
of calling `cleanUp` directly.
- `cleanUpAfterFailure` runs `cleanUp` under the existing system-exit
monitoring. If it throws, it rethrows JVM-fatal errors
(`ExceptionUtils.rethrowIfFatalError`). Otherwise it logs the failure at ERROR
and adds it as suppressed to the original exception (skipped when `cleanUp`
rethrows that same exception).
- The handling of `cancelInvokable` and `transitionStateOnFailure`
failures is unchanged.
## Verifying this change
This change added tests and can be verified as follows:
- `TaskTest#testExceptionInCleanUpOfFailedTaskIsNotFatal`: an invokable
whose `cleanUp` throws while the task is failing. The task ends `FAILED` with
the clean-up exception suppressed on the original failure, and
`notifyFatalError` is not called. This test fails without the fix and passes
with it.
- `TaskTest#testJvmFatalErrorInCleanUpOfFailedTaskIsFatal`: a JVM-fatal
error from `cleanUp` is still reported as fatal. This is a regression guard and
passes both with and without the fix.
- `TaskTest`, `StreamTaskTest` and `SinkV2SinkWriterOperatorTest` pass
locally.
Not run: the full `mvn verify`, checkstyle, and an end-to-end run against
`KafkaWriter`. The new tests use a generic throwing `cleanUp`.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes. It changes how a
failing task handles an exception from its own clean-up, so a task failure no
longer escalates to a TaskManager shutdown.
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Sonnet 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]
