public static RestartBackoffTimeStrategy.Factory createRestartBackoffTimeStrategyFactory(
            final RestartStrategies.RestartStrategyConfiguration jobRestartStrategyConfiguration,
            final Configuration jobConfiguration,
            final Configuration clusterConfiguration,
            final boolean isCheckpointingEnabled) {

        checkNotNull(jobRestartStrategyConfiguration);
        checkNotNull(jobConfiguration);
        checkNotNull(clusterConfiguration);

        return getJobRestartStrategyFactory(jobRestartStrategyConfiguration)
                .orElse(
                        getRestartStrategyFactoryFromConfig(jobConfiguration)
                                .orElse(
                                        (getRestartStrategyFactoryFromConfig(clusterConfiguration)
                                                .orElse(
                                                        getDefaultRestartStrategyFactory(
                                                                isCheckpointingEnabled)))));
    }
@Test
    void testNoRestartStrategySpecifiedInClusterConfig() {
        final Configuration conf = new Configuration();
        conf.set(RestartStrategyOptions.RESTART_STRATEGY, NO_RESTART_STRATEGY.getMainValue());

        final RestartBackoffTimeStrategy.Factory factory =
                RestartBackoffTimeStrategyFactoryLoader.createRestartBackoffTimeStrategyFactory(
                        DEFAULT_JOB_LEVEL_RESTART_CONFIGURATION, new Configuration(), conf, false);

        assertThat(NoRestartBackoffTimeStrategy.NoRestartBackoffTimeStrategyFactory.INSTANCE)
                .isEqualTo(factory);
    }