@VisibleForTesting
    static FlinkSecurityManager fromConfiguration(Configuration configuration) {
        final ClusterOptions.UserSystemExitMode userSystemExitMode =
                configuration.get(ClusterOptions.INTERCEPT_USER_SYSTEM_EXIT);

        boolean haltOnSystemExit = configuration.get(ClusterOptions.HALT_ON_FATAL_ERROR);

        // If no check is needed, return null so that caller can avoid setting security manager not
        // to incur any runtime cost.
        if (userSystemExitMode == ClusterOptions.UserSystemExitMode.DISABLED && !haltOnSystemExit) {
            return null;
        }
        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, haltOnSystemExit);
    }
@Test
    void testDisabledConfiguration() {
        // Default case (no provided option) - allowing everything, so null security manager is
        // expected.
        Configuration configuration = new Configuration();
        FlinkSecurityManager flinkSecurityManager =
                FlinkSecurityManager.fromConfiguration(configuration);
        assertThat(flinkSecurityManager).isNull();

        // Disabled case (same as default)
        configuration.set(
                ClusterOptions.INTERCEPT_USER_SYSTEM_EXIT,
                ClusterOptions.UserSystemExitMode.DISABLED);
        flinkSecurityManager = FlinkSecurityManager.fromConfiguration(configuration);
        assertThat(flinkSecurityManager).isNull();

        // No halt (same as default)
        configuration.set(ClusterOptions.HALT_ON_FATAL_ERROR, false);
        flinkSecurityManager = FlinkSecurityManager.fromConfiguration(configuration);
        assertThat(flinkSecurityManager).isNull();
    }