Enum
cluster.processes.halt-on-fatal-error: In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when THROW is configured.cluster.processes.halt-on-fatal-error: In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when THROW is configured.Enum
cluster.processes.halt-on-fatal-error: In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when THROW is configured.cluster.processes.halt-on-fatal-error: In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when THROW is configured.The Java Security Manager was permanently disabled in JDK 24 (JEP 486). On such JVMs the security manager features + * cannot be enabled, but the static process exit helpers keep working. */ public class FlinkSecurityManager extends SecurityManager { @@ -73,14 +77,13 @@ public class FlinkSecurityManager extends SecurityManager { } /** - * Instantiate FlinkUserSecurityManager from configuration. Return null if no security manager - * check is needed, so that a caller can skip setting security manager avoiding runtime check - * cost, if there is no security check set up already. Use {@link #setFromConfiguration} helper, - * which handles disabled case. + * Instantiate FlinkSecurityManager from configuration. Return null if no security manager check + * is needed, so that a caller can skip setting security manager avoiding runtime check cost, if + * there is no security check set up already. Use {@link #setFromConfiguration} helper, which + * handles disabled case. * * @param configuration to instantiate the security manager from - * @return FlinkUserSecurityManager instantiated based on configuration. Return null if - * disabled. + * @return FlinkSecurityManager instantiated based on configuration. Return null if disabled. */ @VisibleForTesting static FlinkSecurityManager fromConfiguration(Configuration configuration) { @@ -110,12 +113,22 @@ public static void setFromConfiguration(Configuration configuration) { try { System.setSecurityManager(flinkSecurityManager); } catch (Exception e) { + // UnsupportedOperationException: JDK 18-23 without -Djava.security.manager=allow, + // or JDK 24+ where the Security Manager is permanently disabled (JEP 486). + // Anything else: an existing security manager refuses to be replaced. + final String reason = + e instanceof UnsupportedOperationException + ? "the Java Security Manager is not available in this JVM: it was " + + "permanently disabled in JDK 24 (JEP 486) and must be " + + "explicitly enabled with '-Djava.security.manager=allow' " + + "on JDK 18 to 23." + : "there is no permission to set a SecurityManager; update your " + + "existing SecurityManager to allow it."; throw new IllegalConfigurationException( String.format( - "Could not register security manager due to no permission to " - + "set a SecurityManager. Either update your existing " - + "SecurityManager to allow the permission or do not use " - + "security manager features (e.g., '%s: %s', '%s: %s')", + "Could not register security manager because %s Alternatively, " + + "disable the security manager features ('%s: %s', '%s: %s').", + reason, ClusterOptions.INTERCEPT_USER_SYSTEM_EXIT.key(), ClusterOptions.INTERCEPT_USER_SYSTEM_EXIT.defaultValue(), ClusterOptions.HALT_ON_FATAL_ERROR.key(), @@ -182,7 +195,7 @@ public void checkExit(int status) { // System.exit if (haltOnSystemExit) { // null the security manager to prevent infinite recursion - System.setSecurityManager(null); + clearSecurityManager(); Runtime.getRuntime().halt(status); } } @@ -209,11 +222,22 @@ boolean userSystemExitMonitored() { */ public static void forceProcessExit(int exitCode) { // Unset ourselves to allow exiting in any case. - System.setSecurityManager(null); + clearSecurityManager(); if (flinkSecurityManager != null && flinkSecurityManager.haltOnSystemExit) { Runtime.getRuntime().halt(exitCode); } else { System.exit(exitCode); } } + + /** + * Unsets the system security manager, if any. On JDK 24+ {@code System.setSecurityManager} + * always throws (JEP 486), even when passed {@code null}, but there a security manager can + * never be installed in the first place, so there is nothing to unset either. + */ + private static void clearSecurityManager() { + if (System.getSecurityManager() != null) { + System.setSecurityManager(null); + } + } } diff --git a/flink-core/src/main/java/org/apache/flink/util/concurrent/ExecutorThreadFactory.java b/flink-core/src/main/java/org/apache/flink/util/concurrent/ExecutorThreadFactory.java index fe3c496ef6a2da..bc7789776f77c7 100644 --- a/flink-core/src/main/java/org/apache/flink/util/concurrent/ExecutorThreadFactory.java +++ b/flink-core/src/main/java/org/apache/flink/util/concurrent/ExecutorThreadFactory.java @@ -97,11 +97,7 @@ public ExecutorThreadFactory(String poolName, UncaughtExceptionHandler exception this.threadPriority = threadPriority; this.exceptionHandler = exceptionHandler; - SecurityManager securityManager = System.getSecurityManager(); - this.group = - (securityManager != null) - ? securityManager.getThreadGroup() - : Thread.currentThread().getThreadGroup(); + this.group = Thread.currentThread().getThreadGroup(); } // ------------------------------------------------------------------------ diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/ClusterUncaughtExceptionHandlerITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/ClusterUncaughtExceptionHandlerITCase.java index fddfc9bb57676e..fb5457a99868ce 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/ClusterUncaughtExceptionHandlerITCase.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/ClusterUncaughtExceptionHandlerITCase.java @@ -32,7 +32,6 @@ import org.apache.flink.util.concurrent.ScheduledExecutor; import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -42,7 +41,6 @@ import static org.assertj.core.api.Assumptions.assumeThat; /** Integration test to check exit behaviour for the {@link ClusterUncaughtExceptionHandler}. */ -@Tag("org.apache.flink.testutils.junit.FailsOnJava25") @ExtendWith(TestLoggerExtension.class) class ClusterUncaughtExceptionHandlerITCase { 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 e9ca964350e8b7..925a138670522b 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 @@ -110,6 +110,14 @@ public void run() { */ public abstract String getEntryPointClassName(); + /** + * Returns additional arguments for the JVM. They are placed after the arguments derived from + * the surefire configuration, so they take precedence over those. + */ + protected String[] getJvmArgs() { + return new String[0]; + } + // --------------------------------------------------------------------------------------------- /** @@ -149,6 +157,7 @@ public void startProcess() throws IOException { cmd = ArrayUtils.addAll(cmd, moduleConfig.trim().split("\\s+")); } + cmd = ArrayUtils.addAll(cmd, getJvmArgs()); cmd = ArrayUtils.add(cmd, getEntryPointClassName()); String[] mainMethodArgs = getMainMethodArgs(); 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 index 94b2835b4b3424..8826309f1d216d 100644 --- 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 @@ -20,6 +20,7 @@ import org.apache.flink.configuration.ClusterOptions; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.IllegalConfigurationException; import org.apache.flink.core.security.FlinkSecurityManager; import org.apache.flink.runtime.testutils.TestJvmProcess; import org.apache.flink.util.OperatingSystem; @@ -32,9 +33,10 @@ import static org.assertj.core.api.Assumptions.assumeThat; /** Integration tests for the {@link FlinkSecurityManager}. */ -@Tag("org.apache.flink.testutils.junit.FailsOnJava25") class FlinkSecurityManagerITCase { + private static final int FORCED_EXIT_CODE = 222; + @BeforeEach void ensureSupportedOS() { // based on the assumption in JvmExitOnFatalErrorTest, and manual testing on Mac, we do not @@ -42,40 +44,69 @@ void ensureSupportedOS() { assumeThat(OperatingSystem.isLinux() || OperatingSystem.isMac()).isTrue(); } + @Tag("org.apache.flink.testutils.junit.FailsOnJava25") @Test void testForcedJVMExit() throws Exception { - final ForcedJVMExitProcess testProcess = - new ForcedJVMExitProcess(ForcedExitEntryPoint.class); - - testProcess.startProcess(); - try { - testProcess.waitFor(); - assertThat(testProcess.exitCode()).isEqualTo(222); - } finally { - testProcess.destroy(); - } + runAndAssertExitCode( + new ForcedJVMExitProcess(ForcedExitEntryPoint.class), FORCED_EXIT_CODE); } + @Tag("org.apache.flink.testutils.junit.FailsOnJava25") @Test void testIgnoredJVMExit() throws Exception { - final ForcedJVMExitProcess testProcess = - new ForcedJVMExitProcess(IgnoredExitEntryPoint.class); + runAndAssertExitCode(new ForcedJVMExitProcess(IgnoredExitEntryPoint.class), 0); + } + @Test + void testForcedJVMExitWithoutSecurityManager() throws Exception { + runAndAssertExitCode( + new ForcedJVMExitProcess(ForcedExitWithoutSecurityManagerEntryPoint.class), + FORCED_EXIT_CODE); + } + + @Test + void testUnavailableSecurityManagerIsReported() throws Exception { + // 'disallow' is only understood by JDK 12+; JDK 24+ accepts it although the Security + // Manager is disabled there anyway + assumeThat(Runtime.version().feature()).isGreaterThanOrEqualTo(12); + runAndAssertExitCode( + new ForcedJVMExitProcess( + UnavailableSecurityManagerEntryPoint.class, + "-Djava.security.manager=disallow"), + FORCED_EXIT_CODE); + } + + private static void runAndAssertExitCode(ForcedJVMExitProcess testProcess, int expectedExitCode) + throws Exception { testProcess.startProcess(); try { testProcess.waitFor(); - assertThat(testProcess.exitCode()).isZero(); + assertThat(testProcess.exitCode()) + .withFailMessage( + "Unexpected exit code %d. Process output:%n%s", + testProcess.exitCode(), testProcess.getProcessOutput()) + .isEqualTo(expectedExitCode); } finally { testProcess.destroy(); } } + /** Configures the {@link FlinkSecurityManager} to intercept calls to System.exit(). */ + private static void interceptUserSystemExit() { + Configuration configuration = new Configuration(); + configuration.set( + ClusterOptions.INTERCEPT_USER_SYSTEM_EXIT, ClusterOptions.UserSystemExitMode.THROW); + FlinkSecurityManager.setFromConfiguration(configuration); + } + private static final class ForcedJVMExitProcess extends TestJvmProcess { - private final Class entryPointName; + private final Class entryPoint; + private final String[] jvmArgs; - private ForcedJVMExitProcess(Class entryPointName) throws Exception { - this.entryPointName = entryPointName; + private ForcedJVMExitProcess(Class entryPoint, String... jvmArgs) throws Exception { + this.entryPoint = entryPoint; + this.jvmArgs = jvmArgs; } @Override @@ -90,22 +121,22 @@ public String[] getMainMethodArgs() { @Override public String getEntryPointClassName() { - return entryPointName.getName(); + return entryPoint.getName(); + } + + @Override + protected String[] getJvmArgs() { + return jvmArgs; } } /** An entry point that attempts to force exit the process. */ 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); + public static void main(String[] args) { + interceptUserSystemExit(); - FlinkSecurityManager.forceProcessExit(222); + FlinkSecurityManager.forceProcessExit(FORCED_EXIT_CODE); System.exit(0); } @@ -114,13 +145,8 @@ public static void main(String[] args) throws Exception { /** An entry point that attempts to exit the process but expects that call to be ignored. */ 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); + public static void main(String[] args) { + interceptUserSystemExit(); FlinkSecurityManager.monitorUserSystemExitForCurrentThread(); // expect this call to be ignored @@ -132,4 +158,38 @@ public static void main(String[] args) throws Exception { } } } + + /** + * An entry point that force exits the process without any security manager having been + * installed, as is always the case on JDK 24+. + */ + public static final class ForcedExitWithoutSecurityManagerEntryPoint { + + public static void main(String[] args) { + FlinkSecurityManager.forceProcessExit(FORCED_EXIT_CODE); + + System.exit(0); + } + } + + /** + * An entry point that expects enabling the security manager to fail with a descriptive error + * because the JVM does not allow installing one. + */ + public static final class UnavailableSecurityManagerEntryPoint { + + public static void main(String[] args) { + try { + interceptUserSystemExit(); + } catch (IllegalConfigurationException e) { + System.err.println("Caught exception: " + e.getMessage()); + if (e.getMessage().contains("JEP 486") + && e.getCause() instanceof UnsupportedOperationException) { + FlinkSecurityManager.forceProcessExit(FORCED_EXIT_CODE); + } + } + + System.exit(1); + } + } }