[
https://issues.apache.org/jira/browse/FLINK-41269?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18135190#comment-18135190
]
Piotr Nowojski commented on FLINK-41269:
----------------------------------------
As I commented in
https://github.com/apache/flink/pull/26861#discussion_r2247395574, that was
intentional change. If operator fails to close and release it's resources, it
might not be safe to continue running that JVM. This can lead to resource
leaks, or potentialy even more serious problems (some zombie background thread
doing something that conflicts with a restarted subtask).
What is the problem that this is causing? What's the motivation behind this
change? If something doesn't work, because `close` throws an exception, than
that `close` call should be fixed in the operator/function.
> Exception thrown from an operator's `close()` during failure cleanup is
> TaskManager-fatal since FLINK-38180 (it was suppressed before)
> --------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41269
> URL: https://issues.apache.org/jira/browse/FLINK-41269
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: 2.2.0, 2.3.0
> Environment: Flink 2.2.0 on Kubernetes (Flink Kubernetes Operator),
> `flink-connector-kafka` 5.0.0-2.2, Java 21.
> Reporter: Dries Samyn
> Assignee: Dries Samyn
> Priority: Major
> Labels: pull-request-available
>
> Since FLINK-38180 ("Clean up task after switching to FAILED",
> apache/flink#26861), the cleanup of a failing task is registered in
> `Task#postFailureCleanUpRegistry` and run inside the `catch (Throwable t)`
> block of `Task#doRun`:
> {code:java}
> try {
> transitionStateOnFailure(t, postFailureCleanUpRegistry);
> postFailureCleanUpRegistry.close();
> } catch (Throwable tt) {
> // "FATAL - exception in exception handler of task ..."
> notifyFatalError(message, tt);
> }
> {code}
> `restoreAndInvoke` registers `finalInvokable.cleanUp(throwable)` there. Any
> exception thrown by an operator's or sink writer's `close()` while the task
> is already failing therefore reaches `notifyFatalError`, and the TaskManager
> shuts down. All other tasks on that TaskManager fail with it.
> Before the change (checked on the `release-1.20`, `release-2.0` and
> `release-2.1` branches), `restoreAndInvoke` did:
> {code:java}
> } catch (Throwable throwable) {
> try {
> runWithSystemExitMonitoring(() -> finalInvokable.cleanUp(throwable));
> } catch (Throwable cleanUpThrowable) {
> throwable.addSuppressed(cleanUpThrowable);
> }
> throw throwable;
> }
> {code}
> so a failing `close()` was attached to the original failure and the task
> simply failed. The PR description only mentions preventing a race in which a
> cleanup exception hides the real exception; it does not mention that a
> cleanup exception now takes down the TaskManager. If that is intended, the
> release notes and the `Task` Javadoc should say so. If it is not, the fix
> should keep the new ordering but treat a failing cleanup as suppressed.
> `release-2.2` and `master` both contain the new behaviour (as read on GitHub).
> *Concrete trigger seen in the wild*
> `flink-connector-kafka`'s `KafkaWriter#close()` ends with
> `checkAsyncException()`, which rethrows a pending asynchronous producer
> failure. That has been the case since at least connector v3.0.2 (the comment
> "Rethrow exception for the case in which close is called before writer() and
> flush()" is unchanged through v5.0.0 and `main`). With Flink 2.2.0 and
> connector 5.0.0-2.2, a delivery timeout (`Expiring N record(s) for
> <topic>-<partition>: 120000 ms has passed since batch creation`) failed the
> task and `KafkaWriter#close()` then threw `IOException: One or more Kafka
> Producer send requests have encountered exception`. Production logs show
> `FATAL - exception in exception handler of task ...` followed by `Fatal error
> occurred while executing the TaskManager. Shutting it down...` and SIGTERM
> within the same second.
> How the pending exception is still set at `close()` time is my reading of the
> connector source (a later callback in the same expired batch re-arms the
> field after the first one was thrown and cleared).
> *Expected behaviour*
> An exception from `close()` during the cleanup of an already-failing task is
> added as a suppressed exception to the original failure, as in 2.1 and
> earlier. The TaskManager stays up.
> *Actual behaviour*
> The TaskManager shuts down.
> *Proposed change*
> Catch exceptions from `postFailureCleanUpRegistry.close()` separately from
> `transitionStateOnFailure`, add them as suppressed to `t`, and log them,
> instead of passing them to `notifyFatalError`. Keep `notifyFatalError` for a
> failure of `transitionStateOnFailure` itself. Add a `TaskTest` where an
> invokable's `cleanUp` throws while the task is failing and assert that the
> task ends `FAILED` and `notifyFatalError` is not called.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
