[
https://issues.apache.org/jira/browse/FLINK-41269?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134243#comment-18134243
]
Dries Samyn commented on FLINK-41269:
-------------------------------------
Proposed fix: https://github.com/apache/flink/pull/29676
> 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
> 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)
