public static Optional<CheckpointStorage> fromConfig(
            ReadableConfig config, ClassLoader classLoader, @Nullable Logger logger)
            throws IllegalStateException, DynamicCodeLoadingException {

        Preconditions.checkNotNull(config, "config");
        Preconditions.checkNotNull(classLoader, "classLoader");

        final String storageName = config.get(CheckpointingOptions.CHECKPOINT_STORAGE);
        if (storageName == null) {
            if (logger != null) {
                logger.debug(
                        "The configuration {} has not be set in the current"
                                + " sessions config.yaml. Falling back to a default CheckpointStorage"
                                + " type. Users are strongly encouraged explicitly set this configuration"
                                + " so they understand how their applications are checkpointing"
                                + " snapshots for fault-tolerance.",
                        CheckpointingOptions.CHECKPOINT_STORAGE.key());
            }
            return Optional.empty();
        }

        switch (storageName.toLowerCase()) {
            case JOB_MANAGER_STORAGE_NAME:
                return Optional.of(createJobManagerCheckpointStorage(config, classLoader, logger));

            case FILE_SYSTEM_STORAGE_NAME:
                return Optional.of(createFileSystemCheckpointStorage(config, classLoader, logger));

            default:
                if (logger != null) {
                    logger.info("Loading state backend via factory '{}'", storageName);
                }

                CheckpointStorageFactory<?> factory;
                try {
                    @SuppressWarnings("rawtypes")
                    Class<? extends CheckpointStorageFactory> clazz =
                            Class.forName(storageName, false, classLoader)
                                    .asSubclass(CheckpointStorageFactory.class);

                    factory = clazz.newInstance();
                } catch (ClassNotFoundException e) {
                    throw new DynamicCodeLoadingException(
                            "Cannot find configured state backend factory class: " + storageName,
                            e);
                } catch (ClassCastException | InstantiationException | IllegalAccessException e) {
                    throw new DynamicCodeLoadingException(
                            "The class configured under '"
                                    + CheckpointingOptions.CHECKPOINT_STORAGE.key()
                                    + "' is not a valid checkpoint storage factory ("
                                    + storageName
                                    + ')',
                            e);
                }

                return Optional.of(factory.createFromConfig(config, classLoader));
        }
    }
@Test
    void testLoadJobManagerStorageWithParameters() throws Exception {
        final String savepointDir = new Path(TempDirUtils.newFolder(tmp).toURI()).toString();
        final Path expectedSavepointPath = new Path(savepointDir);

        // we configure with the explicit string (rather than
        // AbstractStateBackend#X_STATE_BACKEND_NAME)
        // to guard against config-breaking changes of the name

        final Configuration config1 = new Configuration();
        config1.set(CheckpointingOptions.CHECKPOINT_STORAGE, "jobmanager");
        config1.set(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointDir);

        CheckpointStorage storage1 = CheckpointStorageLoader.fromConfig(config1, cl, null).get();

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

        assertThat(((JobManagerCheckpointStorage) storage1).getSavepointPath())
                .isEqualTo(expectedSavepointPath);
    }