[FLINK-36902][core] Keep process exit working on JDK 24+ where the Security Manager is disabled

docs/layouts/shortcodes/generated/cluster_configuration.html

@@ -18,7 +18,7 @@
             <td><h5>cluster.intercept-user-system-exit</h5></td>
             <td style="word-wrap: break-word;">DISABLED</td>
             <td><p>Enum</p></td>
-            <td>Flag to check user code exiting system by terminating JVM (e.g., System.exit()). Note that this configuration option can interfere with <code class="highlighter-rouge">cluster.processes.halt-on-fatal-error</code>: In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when <code class="highlighter-rouge">THROW</code> is configured.<br /><br />Possible values:<ul><li>"DISABLED": Flink is not monitoring or intercepting calls to System.exit()</li><li>"LOG": Log exit attempt with stack trace but still allowing exit to be performed</li><li>"THROW": Throw exception when exit is attempted disallowing JVM termination</li></ul></td>
+            <td>Flag to check user code exiting system by terminating JVM (e.g., System.exit()). Note that this configuration option can interfere with <code class="highlighter-rouge">cluster.processes.halt-on-fatal-error</code>: In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when <code class="highlighter-rouge">THROW</code> is configured.<br />This option relies on the Java Security Manager, which was permanently disabled in JDK 24 (<a href="https://openjdk.org/jeps/486">JEP 486</a>). On JDK 24 and newer it must be left at its default.<br /><br />Possible values:<ul><li>"DISABLED": Flink is not monitoring or intercepting calls to System.exit()</li><li>"LOG": Log exit attempt with stack trace but still allowing exit to be performed</li><li>"THROW": Throw exception when exit is attempted disallowing JVM termination</li></ul></td>
         </tr>
         <tr>
             <td><h5>cluster.io-pool.size</h5></td>
@@ -30,7 +30,7 @@
             <td><h5>cluster.processes.halt-on-fatal-error</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>Whether processes should halt on fatal errors instead of performing a graceful shutdown. In some environments (e.g. Java 8 with the G1 garbage collector), a regular graceful shutdown can lead to a JVM deadlock. See <a href="https://issues.apache.org/jira/browse/FLINK-16510">FLINK-16510</a> for details.</td>
+            <td>Whether processes should halt on fatal errors instead of performing a graceful shutdown. In some environments (e.g. Java 8 with the G1 garbage collector), a regular graceful shutdown can lead to a JVM deadlock. See <a href="https://issues.apache.org/jira/browse/FLINK-16510">FLINK-16510</a> for details.<br />This option relies on the Java Security Manager, which was permanently disabled in JDK 24 (<a href="https://openjdk.org/jeps/486">JEP 486</a>). On JDK 24 and newer it must be left at its default.</td>
         </tr>
         <tr>
             <td><h5>cluster.registration.error-delay</h5></td>

docs/layouts/shortcodes/generated/expert_cluster_section.html

@@ -18,13 +18,13 @@
             <td><h5>cluster.intercept-user-system-exit</h5></td>
             <td style="word-wrap: break-word;">DISABLED</td>
             <td><p>Enum</p></td>
-            <td>Flag to check user code exiting system by terminating JVM (e.g., System.exit()). Note that this configuration option can interfere with <code class="highlighter-rouge">cluster.processes.halt-on-fatal-error</code>: In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when <code class="highlighter-rouge">THROW</code> is configured.<br /><br />Possible values:<ul><li>"DISABLED": Flink is not monitoring or intercepting calls to System.exit()</li><li>"LOG": Log exit attempt with stack trace but still allowing exit to be performed</li><li>"THROW": Throw exception when exit is attempted disallowing JVM termination</li></ul></td>
+            <td>Flag to check user code exiting system by terminating JVM (e.g., System.exit()). Note that this configuration option can interfere with <code class="highlighter-rouge">cluster.processes.halt-on-fatal-error</code>: In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when <code class="highlighter-rouge">THROW</code> is configured.<br />This option relies on the Java Security Manager, which was permanently disabled in JDK 24 (<a href="https://openjdk.org/jeps/486">JEP 486</a>). On JDK 24 and newer it must be left at its default.<br /><br />Possible values:<ul><li>"DISABLED": Flink is not monitoring or intercepting calls to System.exit()</li><li>"LOG": Log exit attempt with stack trace but still allowing exit to be performed</li><li>"THROW": Throw exception when exit is attempted disallowing JVM termination</li></ul></td>
         </tr>
         <tr>
             <td><h5>cluster.processes.halt-on-fatal-error</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>Whether processes should halt on fatal errors instead of performing a graceful shutdown. In some environments (e.g. Java 8 with the G1 garbage collector), a regular graceful shutdown can lead to a JVM deadlock. See <a href="https://issues.apache.org/jira/browse/FLINK-16510">FLINK-16510</a> for details.</td>
+            <td>Whether processes should halt on fatal errors instead of performing a graceful shutdown. In some environments (e.g. Java 8 with the G1 garbage collector), a regular graceful shutdown can lead to a JVM deadlock. See <a href="https://issues.apache.org/jira/browse/FLINK-16510">FLINK-16510</a> for details.<br />This option relies on the Java Security Manager, which was permanently disabled in JDK 24 (<a href="https://openjdk.org/jeps/486">JEP 486</a>). On JDK 24 and newer it must be left at its default.</td>
         </tr>
         <tr>
             <td><h5>cluster.thread-dump.default-mode</h5></td>

flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java

@@ -115,6 +115,11 @@ public class ClusterOptions {
                                             link(
                                                     "https://issues.apache.org/jira/browse/FLINK-16510",
                                                     "FLINK-16510"))
+                                    .linebreak()
+                                    .text(
+                                            "This option relies on the Java Security Manager, which was permanently disabled in JDK 24 (%s). "
+                                                    + "On JDK 24 and newer it must be left at its default.",
+                                            link("https://openjdk.org/jeps/486", "JEP 486"))
                                     .build());
 
     @Documentation.Section(Documentation.Sections.EXPERT_CLUSTER)
@@ -130,6 +135,11 @@ public class ClusterOptions {
                                             "Note that this configuration option can interfere with %s: "
                                                     + "In intercepted user-code, a call to System.exit() will not cause the JVM to halt, when %s is configured.",
                                             code(HALT_ON_FATAL_ERROR.key()), code(THROW.name()))
+                                    .linebreak()
+                                    .text(
+                                            "This option relies on the Java Security Manager, which was permanently disabled in JDK 24 (%s). "
+                                                    + "On JDK 24 and newer it must be left at its default.",
+                                            link("https://openjdk.org/jeps/486", "JEP 486"))
                                     .build());
 
     @Documentation.Section(Documentation.Sections.EXPERT_CLUSTER)

flink-core/src/main/java/org/apache/flink/core/security/FlinkSecurityManager.java

@@ -37,6 +37,10 @@
  * malicious users who try to tweak security manager on their own, but more for being dependable
  * against user mistakes by gracefully handling them informing users rather than causing silent
  * unavailability.
+ *
+ * <p>The Java Security Manager was permanently disabled in JDK 24 (<a
+ * href="https://openjdk.org/jeps/486">JEP 486</a>). 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);
+        }
+    }
 }

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();
     }
 
     // ------------------------------------------------------------------------

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 {
 

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();