You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2021/08/26 10:54:41 UTC
[flink] 02/06: [hotfix][runtime] Extract
Task.runWithSystemExitMonitoring to simplify subsequent changes
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b139d707bca90f9c3a3c562f4a6f83abbd2bc4af
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Wed Aug 18 18:07:09 2021 +0100
[hotfix][runtime] Extract Task.runWithSystemExitMonitoring to simplify subsequent changes
---
.../org/apache/flink/runtime/taskmanager/Task.java | 37 +++++++++++-----------
1 file changed, 19 insertions(+), 18 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
index bde0ca6..bdb20e2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -87,6 +87,7 @@ import org.apache.flink.util.UserCodeClassLoader;
import org.apache.flink.util.WrappingRuntimeException;
import org.apache.flink.util.clock.SystemClock;
import org.apache.flink.util.concurrent.FutureUtils;
+import org.apache.flink.util.function.RunnableWithException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -760,13 +761,8 @@ public class Task
// make sure the user code classloader is accessible thread-locally
executingThread.setContextClassLoader(userCodeClassLoader.asClassLoader());
- FlinkSecurityManager.monitorUserSystemExitForCurrentThread();
- try {
- // Restore invokable data to the last valid state
- invokable.restore();
- } finally {
- FlinkSecurityManager.unmonitorUserSystemExitForCurrentThread();
- }
+ AbstractInvokable finalInvokable = invokable;
+ runWithSystemExitMonitoring(finalInvokable::restore);
if (!transitionState(ExecutionState.INITIALIZING, ExecutionState.RUNNING)) {
throw new CancelTaskException();
@@ -776,17 +772,7 @@ public class Task
taskManagerActions.updateTaskExecutionState(
new TaskExecutionState(executionId, ExecutionState.RUNNING));
- // Monitor user codes from exiting JVM covering user function invocation. This can be
- // done in a finer-grained way like enclosing user callback functions individually,
- // but as exit triggered by framework is not performed and expected in this invoke
- // function anyhow, we can monitor exiting JVM for entire scope.
- FlinkSecurityManager.monitorUserSystemExitForCurrentThread();
- try {
- // run the invokable
- invokable.invoke();
- } finally {
- FlinkSecurityManager.unmonitorUserSystemExitForCurrentThread();
- }
+ runWithSystemExitMonitoring(finalInvokable::invoke);
// make sure, we enter the catch block if the task leaves the invoke() method due
// to the fact that it has been canceled
@@ -945,6 +931,21 @@ public class Task
}
}
+ /**
+ * Monitor user codes from exiting JVM covering user function invocation. This can be done in a
+ * finer-grained way like enclosing user callback functions individually, but as exit triggered
+ * by framework is not performed and expected in this invoke function anyhow, we can monitor
+ * exiting JVM for entire scope.
+ */
+ private void runWithSystemExitMonitoring(RunnableWithException action) throws Exception {
+ FlinkSecurityManager.monitorUserSystemExitForCurrentThread();
+ try {
+ action.run();
+ } finally {
+ FlinkSecurityManager.unmonitorUserSystemExitForCurrentThread();
+ }
+ }
+
@VisibleForTesting
public static void setupPartitionsAndGates(
ResultPartitionWriter[] producedPartitions, InputGate[] inputGates) throws IOException {