[
https://issues.apache.org/jira/browse/FLINK-41269?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dries Samyn updated FLINK-41269:
--------------------------------
Description:
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.
was:
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:java}
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.
> 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
> Priority: Major
>
> 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)
