public static void setFromConfiguration(Configuration configuration) {
        final FlinkSecurityManager flinkSecurityManager =
                FlinkSecurityManager.fromConfiguration(configuration);
        if (flinkSecurityManager != null) {
            try {
                System.setSecurityManager(flinkSecurityManager);
            } catch (Exception e) {
                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')",
                                ClusterOptions.INTERCEPT_USER_SYSTEM_EXIT.key(),
                                ClusterOptions.INTERCEPT_USER_SYSTEM_EXIT.defaultValue(),
                                ClusterOptions.HALT_ON_FATAL_ERROR.key(),
                                ClusterOptions.HALT_ON_FATAL_ERROR.defaultValue()),
                        e);
            }
        }
        FlinkSecurityManager.flinkSecurityManager = flinkSecurityManager;
    }
@Test
    void testRegistrationNotAllowedByExistingSecurityManager() {
        Configuration configuration = new Configuration();
        configuration.set(
                ClusterOptions.INTERCEPT_USER_SYSTEM_EXIT, ClusterOptions.UserSystemExitMode.THROW);

        System.setSecurityManager(
                new SecurityManager() {

                    private boolean fired;

                    @Override
                    public void checkPermission(Permission perm) {
                        if (!fired && perm.getName().equals("setSecurityManager")) {
                            try {
                                throw new SecurityException("not allowed");
                            } finally {
                                // Allow removing this manager again
                                fired = true;
                            }
                        }
                    }
                });

        assertThatThrownBy(() -> FlinkSecurityManager.setFromConfiguration(configuration))
                .isInstanceOf(IllegalConfigurationException.class)
                .hasMessageContaining("Could not register security manager");
    }