[GitHub] [flink] hwanju commented on a change in pull request #14499: [FLINK-15156] Warn user if System.exit() is called in user code

[GitBox <gi...@apache.org>]

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:
users@infra.apache.org