You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2021/03/30 06:23:13 UTC
[flink] branch master updated: [FLINK-21306] Ensure fatal errors
always stop the JVM process
This is an automated email from the ASF dual-hosted git repository.
rmetzger pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 4b21391 [FLINK-21306] Ensure fatal errors always stop the JVM process
4b21391 is described below
commit 4b2139189e2c98018c77d25b919ca1f11dbf4a0d
Author: Robert Metzger <rm...@apache.org>
AuthorDate: Sat Mar 6 16:17:52 2021 +0100
[FLINK-21306] Ensure fatal errors always stop the JVM process
This closes #15108
---
.../runtime/entrypoint/ClusterEntrypoint.java | 2 +-
.../runtime/security/FlinkSecurityManager.java | 60 +++++-----
.../runtime/taskexecutor/TaskManagerRunner.java | 2 +-
.../runtime/util/FatalExitExceptionHandler.java | 4 +-
.../runtime/security/FlinkSecurityManagerTest.java | 50 ++------
.../flink/runtime/testutils/TestJvmProcess.java | 9 ++
.../runtime/util/FlinkSecurityManagerITCase.java | 133 +++++++++++++++++++++
.../runtime/util/JvmExitOnFatalErrorTest.java | 3 +-
8 files changed, 185 insertions(+), 78 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
index 04111ac..aab1c7a 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -444,7 +444,7 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro
ClusterEntryPointExceptionUtils.tryEnrichClusterEntryPointError(exception);
LOG.error("Fatal error occurred in the cluster entrypoint.", exception);
- System.exit(RUNTIME_FAILURE_RETURN_CODE);
+ FlinkSecurityManager.forceProcessExit(RUNTIME_FAILURE_RETURN_CODE);
}
// --------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/security/FlinkSecurityManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/security/FlinkSecurityManager.java
index 789322b..d543c45 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/security/FlinkSecurityManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/security/FlinkSecurityManager.java
@@ -28,10 +28,7 @@ import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.Nullable;
-
import java.security.Permission;
-import java.util.function.Consumer;
/**
* {@code FlinkSecurityManager} to control certain behaviors that can be captured by Java system
@@ -58,23 +55,21 @@ public class FlinkSecurityManager extends SecurityManager {
private final ThreadLocal<Boolean> monitorUserSystemExit = new InheritableThreadLocal<>();
private final ClusterOptions.UserSystemExitMode userSystemExitMode;
- /** The behavior to execute when the JVM exists. */
- private final Consumer<Integer> onExitBehavior;
+ private final boolean haltOnSystemExit;
@VisibleForTesting
FlinkSecurityManager(
- ClusterOptions.UserSystemExitMode userSystemExitMode,
- @Nullable Consumer<Integer> onExitBehavior) {
- this(userSystemExitMode, onExitBehavior, System.getSecurityManager());
+ ClusterOptions.UserSystemExitMode userSystemExitMode, boolean haltOnSystemExit) {
+ this(userSystemExitMode, haltOnSystemExit, System.getSecurityManager());
}
@VisibleForTesting
FlinkSecurityManager(
ClusterOptions.UserSystemExitMode userSystemExitMode,
- @Nullable Consumer<Integer> onExitBehavior,
+ boolean haltOnSystemExit,
SecurityManager originalSecurityManager) {
this.userSystemExitMode = Preconditions.checkNotNull(userSystemExitMode);
- this.onExitBehavior = onExitBehavior;
+ this.haltOnSystemExit = haltOnSystemExit;
this.originalSecurityManager = originalSecurityManager;
}
@@ -100,22 +95,13 @@ public class FlinkSecurityManager extends SecurityManager {
if (userSystemExitMode == ClusterOptions.UserSystemExitMode.DISABLED && !haltOnSystemExit) {
return null;
}
- Consumer<Integer> onExitBehavior = null;
- // If halt on system exit is configured, registers a custom SecurityManager which converts
- // graceful exists calls using {@code System#exit} into forceful exit calls using
- // {@code Runtime#halt}. The latter does not perform a clean shutdown using the registered
- // shutdown hooks. This may be configured to prevent deadlocks with Java 8 and the G1
- // garbage collection, see https://issues.apache.org/jira/browse/FLINK-16510.
- if (haltOnSystemExit) {
- onExitBehavior = status -> Runtime.getRuntime().halt(status);
- }
LOG.info(
"FlinkSecurityManager is created with {} user system exit mode and {} exit",
userSystemExitMode,
haltOnSystemExit ? "forceful" : "graceful");
// Add more configuration parameters that need user security manager (currently only for
// system exit).
- return new FlinkSecurityManager(userSystemExitMode, onExitBehavior);
+ return new FlinkSecurityManager(userSystemExitMode, haltOnSystemExit);
}
public static void setFromConfiguration(Configuration configuration) {
@@ -142,14 +128,14 @@ public class FlinkSecurityManager extends SecurityManager {
}
public static void monitorUserSystemExitForCurrentThread() {
- if (FlinkSecurityManager.flinkSecurityManager != null) {
- FlinkSecurityManager.flinkSecurityManager.monitorUserSystemExit();
+ if (flinkSecurityManager != null) {
+ flinkSecurityManager.monitorUserSystemExit();
}
}
public static void unmonitorUserSystemExitForCurrentThread() {
- if (FlinkSecurityManager.flinkSecurityManager != null) {
- FlinkSecurityManager.flinkSecurityManager.unmonitorUserSystemExit();
+ if (flinkSecurityManager != null) {
+ flinkSecurityManager.unmonitorUserSystemExit();
}
}
@@ -193,13 +179,10 @@ public class FlinkSecurityManager extends SecurityManager {
if (originalSecurityManager != null) {
originalSecurityManager.checkExit(status);
}
- // At this point, exit is determined. Invoke exit handler if defined, otherwise check ended.
- if (onExitBehavior != null) {
- // Unset ourselves to allow exiting.
- System.setSecurityManager(null);
- // Execute the desired behavior, e.g. forceful exit instead of proceeding with
- // System.exit().
- onExitBehavior.accept(status);
+ // At this point, exit is determined. Halt if defined, otherwise check ended, JVM will call
+ // System.exit
+ if (haltOnSystemExit) {
+ Runtime.getRuntime().halt(status);
}
}
@@ -217,4 +200,19 @@ public class FlinkSecurityManager extends SecurityManager {
boolean userSystemExitMonitored() {
return Boolean.TRUE.equals(monitorUserSystemExit.get());
}
+
+ /**
+ * Use this method to circumvent the configured {@link FlinkSecurityManager} behavior, ensuring
+ * that the current JVM process will always stop via System.exit() or
+ * Runtime.getRuntime().halt().
+ */
+ public static void forceProcessExit(int exitCode) {
+ // Unset ourselves to allow exiting in any case.
+ System.setSecurityManager(null);
+ if (flinkSecurityManager != null && flinkSecurityManager.haltOnSystemExit) {
+ Runtime.getRuntime().halt(exitCode);
+ } else {
+ System.exit(exitCode);
+ }
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index 61b267c..3848427 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -332,7 +332,7 @@ public class TaskManagerRunner implements FatalErrorHandler {
}
private void terminateJVM() {
- System.exit(FAILURE_EXIT_CODE);
+ FlinkSecurityManager.forceProcessExit(FAILURE_EXIT_CODE);
}
// --------------------------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/FatalExitExceptionHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/FatalExitExceptionHandler.java
index a7bafc8..130ef89 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/FatalExitExceptionHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/FatalExitExceptionHandler.java
@@ -18,6 +18,8 @@
package org.apache.flink.runtime.util;
+import org.apache.flink.runtime.security.FlinkSecurityManager;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,7 +45,7 @@ public final class FatalExitExceptionHandler implements Thread.UncaughtException
t.getName(),
e);
} finally {
- System.exit(EXIT_CODE);
+ FlinkSecurityManager.forceProcessExit(EXIT_CODE);
}
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/security/FlinkSecurityManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/security/FlinkSecurityManagerTest.java
index 69d6ebf..dcc693e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/security/FlinkSecurityManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/security/FlinkSecurityManagerTest.java
@@ -26,7 +26,6 @@ import org.apache.flink.runtime.UserSystemExitException;
import org.apache.flink.util.TestLogger;
import org.junit.After;
-import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -34,7 +33,6 @@ import java.security.Permission;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.flink.core.testutils.CommonTestUtils.assertThrows;
import static org.hamcrest.Matchers.is;
@@ -64,7 +62,7 @@ public class FlinkSecurityManagerTest extends TestLogger {
@Test(expected = UserSystemExitException.class)
public void testThrowUserExit() {
FlinkSecurityManager flinkSecurityManager =
- new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.THROW, null);
+ new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.THROW, false);
flinkSecurityManager.monitorUserSystemExit();
flinkSecurityManager.checkExit(TEST_EXIT_CODE);
}
@@ -72,7 +70,7 @@ public class FlinkSecurityManagerTest extends TestLogger {
@Test
public void testToggleUserExit() {
FlinkSecurityManager flinkSecurityManager =
- new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.THROW, null);
+ new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.THROW, false);
flinkSecurityManager.checkExit(TEST_EXIT_CODE);
flinkSecurityManager.monitorUserSystemExit();
try {
@@ -87,7 +85,7 @@ public class FlinkSecurityManagerTest extends TestLogger {
@Test
public void testPerThreadThrowUserExit() throws Exception {
FlinkSecurityManager flinkSecurityManager =
- new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.THROW, null);
+ new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.THROW, false);
ExecutorService executorService = Executors.newSingleThreadExecutor();
// Async thread test before enabling monitoring ensures it does not throw while prestarting
// worker thread, which is to be unmonitored and tested after enabling monitoring enabled.
@@ -111,7 +109,7 @@ public class FlinkSecurityManagerTest extends TestLogger {
@Test
public void testInheritedThrowUserExit() throws Exception {
FlinkSecurityManager flinkSecurityManager =
- new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.THROW, null);
+ new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.THROW, false);
flinkSecurityManager.monitorUserSystemExit();
try {
flinkSecurityManager.checkExit(TEST_EXIT_CODE);
@@ -140,39 +138,7 @@ public class FlinkSecurityManagerTest extends TestLogger {
// Log mode enables monitor but only logging allowing exit, hence not expecting exception.
// NOTE - Do not specifically test warning logging.
FlinkSecurityManager flinkSecurityManager =
- new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.LOG, null);
- flinkSecurityManager.monitorUserSystemExit();
- flinkSecurityManager.checkExit(TEST_EXIT_CODE);
- }
-
- @Test
- public void testExitHandlerTriggered() {
- AtomicInteger exitStatus = new AtomicInteger(0);
- FlinkSecurityManager flinkSecurityManager =
- new FlinkSecurityManager(
- ClusterOptions.UserSystemExitMode.DISABLED, exitStatus::set);
- flinkSecurityManager.checkExit(TEST_EXIT_CODE);
- assertThat(exitStatus.get(), is(TEST_EXIT_CODE));
- }
-
- @Test
- public void testExitHandlerTriggeredWithExistingSecurityManager() {
- TestExitSecurityManager existingSecurityManager = new TestExitSecurityManager();
- System.setSecurityManager(existingSecurityManager);
- AtomicInteger customExitExecuted = new AtomicInteger(0);
- FlinkSecurityManager flinkSecurityManager =
- new FlinkSecurityManager(
- ClusterOptions.UserSystemExitMode.DISABLED, customExitExecuted::set);
-
- flinkSecurityManager.checkExit(TEST_EXIT_CODE);
- assertThat(existingSecurityManager.getExitStatus(), is(TEST_EXIT_CODE));
- assertThat(customExitExecuted.get(), is(TEST_EXIT_CODE));
- }
-
- @Test(expected = UserSystemExitException.class)
- public void testExitBehaviorUnchangeOnThrowingUserExit() {
- FlinkSecurityManager flinkSecurityManager =
- new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.THROW, status -> fail());
+ new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.LOG, false);
flinkSecurityManager.monitorUserSystemExit();
flinkSecurityManager.checkExit(TEST_EXIT_CODE);
}
@@ -240,7 +206,7 @@ public class FlinkSecurityManagerTest extends TestLogger {
// since currently
// there is only one option to have a valid security manager, so test with constructor).
flinkSecurityManager =
- new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.DISABLED, null);
+ new FlinkSecurityManager(ClusterOptions.UserSystemExitMode.DISABLED, false);
flinkSecurityManager.monitorUserSystemExit();
assertTrue(flinkSecurityManager.userSystemExitMonitored());
flinkSecurityManager.checkExit(TEST_EXIT_CODE);
@@ -276,9 +242,7 @@ public class FlinkSecurityManagerTest extends TestLogger {
};
FlinkSecurityManager flinkSecurityManager =
new FlinkSecurityManager(
- ClusterOptions.UserSystemExitMode.DISABLED,
- status -> Assert.fail(),
- originalSecurityManager);
+ ClusterOptions.UserSystemExitMode.DISABLED, false, originalSecurityManager);
assertThrows(
"not allowed",
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestJvmProcess.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestJvmProcess.java
index 7926641..65e2c33 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestJvmProcess.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestJvmProcess.java
@@ -288,6 +288,15 @@ public abstract class TestJvmProcess {
}
}
+ public int exitCode() {
+ Process process = this.process;
+ if (process != null) {
+ return process.exitValue();
+ } else {
+ throw new IllegalStateException("process not started");
+ }
+ }
+
// ---------------------------------------------------------------------------------------------
// File based synchronization utilities
// ---------------------------------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/FlinkSecurityManagerITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/FlinkSecurityManagerITCase.java
new file mode 100644
index 0000000..14960b0
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/FlinkSecurityManagerITCase.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.util;
+
+import org.apache.flink.configuration.ClusterOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.security.FlinkSecurityManager;
+import org.apache.flink.runtime.testutils.TestJvmProcess;
+import org.apache.flink.util.OperatingSystem;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assume.assumeTrue;
+
+/** Integration tests for the {@link FlinkSecurityManager}. */
+public class FlinkSecurityManagerITCase extends TestLogger {
+
+ @Before
+ public void ensureSupportedOS() {
+ // based on the assumption in JvmExitOnFatalErrorTest, and manual testing on Mac, we do not
+ // support all platforms (in particular not Windows)
+ assumeTrue(OperatingSystem.isLinux() || OperatingSystem.isMac());
+ }
+
+ @Test
+ public void testForcedJVMExit() throws Exception {
+ final ForcedJVMExitProcess testProcess =
+ new ForcedJVMExitProcess(ForcedExitEntryPoint.class);
+
+ testProcess.startProcess();
+ try {
+ testProcess.waitFor();
+ assertThat(testProcess.exitCode(), is(222));
+ } finally {
+ testProcess.destroy();
+ }
+ }
+
+ @Test
+ public void testIgnoredJVMExit() throws Exception {
+ final ForcedJVMExitProcess testProcess =
+ new ForcedJVMExitProcess(IgnoredExitEntryPoint.class);
+
+ testProcess.startProcess();
+ try {
+ testProcess.waitFor();
+ assertThat(testProcess.exitCode(), is(0));
+ } finally {
+ testProcess.destroy();
+ }
+ }
+
+ private static final class ForcedJVMExitProcess extends TestJvmProcess {
+
+ private final Class<?> entryPointName;
+
+ private ForcedJVMExitProcess(Class<?> entryPointName) throws Exception {
+ this.entryPointName = entryPointName;
+ }
+
+ @Override
+ public String getName() {
+ return getEntryPointClassName();
+ }
+
+ @Override
+ public String[] getJvmArgs() {
+ return new String[0];
+ }
+
+ @Override
+ public String getEntryPointClassName() {
+ return entryPointName.getName();
+ }
+ }
+
+ public static final class ForcedExitEntryPoint {
+
+ public static void main(String[] args) throws Exception {
+ Configuration configuration = new Configuration();
+ // configure FlinkSecurityManager to intercept calls to System.exit().
+ configuration.set(
+ ClusterOptions.INTERCEPT_USER_SYSTEM_EXIT,
+ ClusterOptions.UserSystemExitMode.THROW);
+ FlinkSecurityManager.setFromConfiguration(configuration);
+
+ FlinkSecurityManager.forceProcessExit(222);
+
+ System.exit(0);
+ }
+ }
+
+ public static final class IgnoredExitEntryPoint {
+
+ public static void main(String[] args) throws Exception {
+ Configuration configuration = new Configuration();
+ // configure FlinkSecurityManager to intercept calls to System.exit().
+ configuration.set(
+ ClusterOptions.INTERCEPT_USER_SYSTEM_EXIT,
+ ClusterOptions.UserSystemExitMode.THROW);
+ FlinkSecurityManager.setFromConfiguration(configuration);
+
+ FlinkSecurityManager.monitorUserSystemExitForCurrentThread();
+ // expect this call to be ignored
+ try {
+ System.exit(123);
+ } catch (Throwable t) {
+ System.err.println(
+ "Caught exception during system exit with message: " + t.getMessage());
+ }
+ }
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java
index fce705c..7e84d19 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java
@@ -68,6 +68,7 @@ import org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo;
import org.apache.flink.runtime.testutils.TestJvmProcess;
import org.apache.flink.util.OperatingSystem;
import org.apache.flink.util.SerializedValue;
+import org.apache.flink.util.TestLogger;
import org.junit.Rule;
import org.junit.Test;
@@ -85,7 +86,7 @@ import static org.mockito.Mockito.mock;
* Test that verifies the behavior of blocking shutdown hooks and of the {@link
* JvmShutdownSafeguard} that guards against it.
*/
-public class JvmExitOnFatalErrorTest {
+public class JvmExitOnFatalErrorTest extends TestLogger {
@Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();