hwanju commented on a change in pull request #14499:
URL: https://github.com/apache/flink/pull/14499#discussion_r549123415
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -527,38 +528,48 @@ protected void beforeInvoke() throws Exception {
@Override
public final void invoke() throws Exception {
+ // Monitor user codes from exiting JVM. This can be done in a
finer-grained way like enclosing user callback
+ // functions individually (open, close, invoke, run, cancel,
etc), but as Flink-managed exit is not performed
+ // and expected in this invoke function anyhow, we can monitor
exiting JVM for entire scope. Cancel and
+ // checkpoint are handled in their own methods.
+ FlinkUserSecurityManager.monitorSystemExitForCurrentThread();
try {
- beforeInvoke();
+ try {
+ beforeInvoke();
- // final check to exit early before starting to run
- if (canceled) {
- throw new CancelTaskException();
- }
+ // final check to exit early before starting to
run
+ if (canceled) {
+ throw new CancelTaskException();
+ }
- // let the task do its work
- runMailboxLoop();
+ // let the task do its work
+ runMailboxLoop();
- // if this left the run() method cleanly despite the
fact that this was canceled,
- // make sure the "clean shutdown" is not attempted
- if (canceled) {
- throw new CancelTaskException();
- }
+ // if this left the run() method cleanly
despite the fact that this was canceled,
+ // make sure the "clean shutdown" is not
attempted
+ if (canceled) {
+ throw new CancelTaskException();
+ }
- afterInvoke();
- }
- catch (Throwable invokeException) {
- failing = !canceled;
- try {
- cleanUpInvoke();
- }
- // TODO: investigate why Throwable instead of Exception
is used here.
- catch (Throwable cleanUpException) {
- Throwable throwable =
ExceptionUtils.firstOrSuppressed(cleanUpException, invokeException);
- ExceptionUtils.rethrowException(throwable);
+ afterInvoke();
+ } catch (Throwable invokeException) {
+ failing = !canceled;
+ try {
+ cleanUpInvoke();
+ }
+ // TODO: investigate why Throwable instead of
Exception is used here.
+ catch (Throwable cleanUpException) {
+ Throwable throwable =
ExceptionUtils.firstOrSuppressed(
+ cleanUpException,
+ invokeException);
+
ExceptionUtils.rethrowException(throwable);
+ }
+
ExceptionUtils.rethrowException(invokeException);
}
- ExceptionUtils.rethrowException(invokeException);
+ cleanUpInvoke();
+ } finally {
+
FlinkUserSecurityManager.unmonitorSystemExitForCurrentThread();
Review comment:
Adding some explanation here, we started with `StreamTask.invoke`
protection. Before in 1.6/1.8, the patch was clean but now with the latest
version of invoke, patch isn't clean much like this. The first try (to avoid
indentation) was to simply add finally at the end with existing try/catch
block, but we also would like to protect against `cleanUpInvoke`, where
user-defined dispose/close is called. The second is making this existing body
as invokeInternal, which is then wrapped with monitor system exit to avoid
indentation in the patch. I didn't yet do that as that's still a little
intrusive. The third option is to protect all invoke() at generic caller level
(i.e., `invokable.invoke()` in `Task.doRun()`) not just for StreamTask. But
there is some another caller of invoke other than Task.doRun() - currently
`BoundedOneInputStreamTaskRunner.mapPartition` (but maybe missing this could be
fine in the first version). I actually became preferring the third option as
it's cleaner an
d covers not just StreamTask, but I am leaving this as is currently to hear
about any opinion.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
