public static CheckpointStorage load(
            @Nullable CheckpointStorage fromApplication,
            StateBackend configuredStateBackend,
            Configuration jobConfig,
            Configuration clusterConfig,
            ClassLoader classLoader,
            @Nullable Logger logger)
            throws IllegalConfigurationException, DynamicCodeLoadingException {

        Preconditions.checkNotNull(jobConfig, "jobConfig");
        Preconditions.checkNotNull(clusterConfig, "clusterConfig");
        Preconditions.checkNotNull(classLoader, "classLoader");
        Preconditions.checkNotNull(configuredStateBackend, "statebackend");

        // Job level config can override the cluster level config.
        Configuration mergedConfig = new Configuration(clusterConfig);
        mergedConfig.addAll(jobConfig);

        // Legacy state backends always take precedence for backwards compatibility.
        StateBackend rootStateBackend =
                (configuredStateBackend instanceof DelegatingStateBackend)
                        ? ((DelegatingStateBackend) configuredStateBackend)
                                .getDelegatedStateBackend()
                        : configuredStateBackend;

        if (rootStateBackend instanceof CheckpointStorage) {
            if (logger != null) {
                logger.info(
                        "Using legacy state backend {} as Job checkpoint storage",
                        rootStateBackend);
                if (fromApplication != null) {
                    logger.warn(
                            "Checkpoint storage passed via StreamExecutionEnvironment is ignored because legacy state backend '{}' is used. {}",
                            rootStateBackend.getClass().getName(),
                            LEGACY_PRECEDENCE_LOG_MESSAGE);
                }
                if (mergedConfig.get(CheckpointingOptions.CHECKPOINT_STORAGE) != null) {
                    logger.warn(
                            "Config option '{}' is ignored because legacy state backend '{}' is used. {}",
                            CheckpointingOptions.CHECKPOINT_STORAGE.key(),
                            rootStateBackend.getClass().getName(),
                            LEGACY_PRECEDENCE_LOG_MESSAGE);
                }
            }
            return (CheckpointStorage) rootStateBackend;
        }

        // In the FLINK-2.0, the checkpoint storage from application will not be supported
        // anymore.
        if (fromApplication != null) {
            if (fromApplication instanceof ConfigurableCheckpointStorage) {
                if (logger != null) {
                    logger.info(
                            "Using job/cluster config to configure application-defined checkpoint storage: {}",
                            fromApplication);
                    if (mergedConfig.get(CheckpointingOptions.CHECKPOINT_STORAGE) != null) {
                        logger.warn(
                                "Config option '{}' is ignored because the checkpoint storage passed via StreamExecutionEnvironment takes precedence.",
                                CheckpointingOptions.CHECKPOINT_STORAGE.key());
                    }
                }
                return ((ConfigurableCheckpointStorage) fromApplication)
                        // Use cluster config for backwards compatibility.
                        .configure(clusterConfig, classLoader);
            }
            if (logger != null) {
                logger.info("Using application defined checkpoint storage: {}", fromApplication);
            }
            return fromApplication;
        }

        return fromConfig(mergedConfig, classLoader, logger)
                .orElseGet(() -> createDefaultCheckpointStorage(mergedConfig, classLoader, logger));
    }
@Test
    void testConfigureJobManagerStorage() throws Exception {
        final String savepointDir = new Path(TempDirUtils.newFolder(tmp).toURI()).toString();
        final Path expectedSavepointPath = new Path(savepointDir);

        final int maxSize = 100;

        final Configuration config = new Configuration();
        config.set(
                CheckpointingOptions.CHECKPOINT_STORAGE,
                "filesystem"); // check that this is not accidentally picked up
        config.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointDir);

        CheckpointStorage storage1 =
                CheckpointStorageLoader.load(
                        new JobManagerCheckpointStorage(maxSize),
                        new ModernStateBackend(),
                        new Configuration(),
                        config,
                        cl,
                        LOG);

        assertThat(storage1).isInstanceOf(JobManagerCheckpointStorage.class);
        JobManagerCheckpointStorage jmStorage = (JobManagerCheckpointStorage) storage1;

        assertThat(jmStorage.getSavepointPath())
                .is(matching(normalizedPath(expectedSavepointPath)));
        assertThat(jmStorage.getMaxStateSize()).isEqualTo(maxSize);
    }