MartijnVisser commented on code in PR #29676:
URL: https://github.com/apache/flink/pull/29676#discussion_r4423782503
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java:
##########
@@ -976,13 +976,35 @@ private void restoreAndInvoke(
runWithSystemExitMonitoring(finalInvokable::invoke);
} catch (Throwable throwable) {
- cleanUpRegistry.registerCloseable(
- () -> runWithSystemExitMonitoring(() ->
finalInvokable.cleanUp(throwable)));
+ cleanUpRegistry.registerCloseable(() ->
cleanUpAfterFailure(finalInvokable, throwable));
throw throwable;
}
runWithSystemExitMonitoring(() -> finalInvokable.cleanUp(null));
}
+ /**
+ * Cleans up the invokable after it failed with the given {@code failure}.
The cleanup runs
+ * after the task switched to its failed state, and the failure was
already reported at that
Review Comment:
The final state only reaches the JobManager in `notifyFinalState`, after
this runs. That is also why the suppressed exception shows up in the exception
history. Please drop "the failure was already reported".
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java:
##########
@@ -976,13 +976,35 @@ private void restoreAndInvoke(
runWithSystemExitMonitoring(finalInvokable::invoke);
} catch (Throwable throwable) {
- cleanUpRegistry.registerCloseable(
- () -> runWithSystemExitMonitoring(() ->
finalInvokable.cleanUp(throwable)));
+ cleanUpRegistry.registerCloseable(() ->
cleanUpAfterFailure(finalInvokable, throwable));
throw throwable;
}
runWithSystemExitMonitoring(() -> finalInvokable.cleanUp(null));
}
+ /**
+ * Cleans up the invokable after it failed with the given {@code failure}.
The cleanup runs
+ * after the task switched to its failed state, and the failure was
already reported at that
+ * point. A non-fatal exception from the cleanup is therefore logged and
added as suppressed to
+ * {@code failure} instead of being propagated, as it would otherwise be
treated as a fatal
+ * error of the TaskManager.
+ */
+ private void cleanUpAfterFailure(TaskInvokable invokable, Throwable
failure) throws Exception {
Review Comment:
`TaskInvokable#cleanUp` doesn't say what happens when it throws. Can you
document there that an exception after a failure is added to that failure as
suppressed, and that only JVM-fatal errors are fatal?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java:
##########
@@ -976,13 +976,35 @@ private void restoreAndInvoke(
runWithSystemExitMonitoring(finalInvokable::invoke);
} catch (Throwable throwable) {
- cleanUpRegistry.registerCloseable(
- () -> runWithSystemExitMonitoring(() ->
finalInvokable.cleanUp(throwable)));
+ cleanUpRegistry.registerCloseable(() ->
cleanUpAfterFailure(finalInvokable, throwable));
throw throwable;
}
runWithSystemExitMonitoring(() -> finalInvokable.cleanUp(null));
}
+ /**
+ * Cleans up the invokable after it failed with the given {@code failure}.
The cleanup runs
+ * after the task switched to its failed state, and the failure was
already reported at that
+ * point. A non-fatal exception from the cleanup is therefore logged and
added as suppressed to
+ * {@code failure} instead of being propagated, as it would otherwise be
treated as a fatal
+ * error of the TaskManager.
+ */
+ private void cleanUpAfterFailure(TaskInvokable invokable, Throwable
failure) throws Exception {
+ try {
+ runWithSystemExitMonitoring(() -> invokable.cleanUp(failure));
+ } catch (Throwable cleanUpFailure) {
+ ExceptionUtils.rethrowIfFatalError(cleanUpFailure);
+ LOG.error(
+ "Error while cleaning up failed task {} ({}).",
+ taskNameWithSubtask,
+ executionId,
+ cleanUpFailure);
+ if (cleanUpFailure != failure) {
Review Comment:
Nothing tests this check. Without it, a `cleanUp` that rethrows the failure
it was given hits "Self-suppression not permitted" and takes the TaskManager
down again. Can you add a test for that?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java:
##########
@@ -976,13 +976,35 @@ private void restoreAndInvoke(
runWithSystemExitMonitoring(finalInvokable::invoke);
} catch (Throwable throwable) {
- cleanUpRegistry.registerCloseable(
- () -> runWithSystemExitMonitoring(() ->
finalInvokable.cleanUp(throwable)));
+ cleanUpRegistry.registerCloseable(() ->
cleanUpAfterFailure(finalInvokable, throwable));
throw throwable;
}
runWithSystemExitMonitoring(() -> finalInvokable.cleanUp(null));
}
+ /**
+ * Cleans up the invokable after it failed with the given {@code failure}.
The cleanup runs
+ * after the task switched to its failed state, and the failure was
already reported at that
+ * point. A non-fatal exception from the cleanup is therefore logged and
added as suppressed to
+ * {@code failure} instead of being propagated, as it would otherwise be
treated as a fatal
+ * error of the TaskManager.
+ */
+ private void cleanUpAfterFailure(TaskInvokable invokable, Throwable
failure) throws Exception {
+ try {
+ runWithSystemExitMonitoring(() -> invokable.cleanUp(failure));
+ } catch (Throwable cleanUpFailure) {
+ ExceptionUtils.rethrowIfFatalError(cleanUpFailure);
Review Comment:
An `OutOfMemoryError` from `cleanUp` is now only logged, while an OOM from
`invoke` halts the JVM when `taskmanager.jvm-exit-on-oom` is set. Is that
intended?
--
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]
