Dries Samyn created FLINK-41269:
-----------------------------------
Summary: 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
Environment: Flink 2.2.0 on Kubernetes (Flink Kubernetes Operator),
`flink-connector-kafka` 5.0.0-2.2, Java 21.
Reporter: Dries Samyn
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`:
```java
try {
transitionStateOnFailure(t, postFailureCleanUpRegistry);
postFailureCleanUpRegistry.close();
} catch (Throwable tt) {
// "FATAL - exception in exception handler of task ..."
notifyFatalError(message, tt);
}
```
`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:
```java
} catch (Throwable throwable) {
try {
runWithSystemExitMonitoring(() -> finalInvokable.cleanUp(throwable));
} catch (Throwable cleanUpThrowable) {
throwable.addSuppressed(cleanUpThrowable);
}
throw throwable;
}
```
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)
