@Override
    public CheckpointStreamFactory resolveCheckpointStorageLocation(
            long checkpointId, CheckpointStorageLocationReference reference) throws IOException {

        if (reference.isDefaultReference()) {
            // default reference, construct the default location for that particular checkpoint
            final Path checkpointDir =
                    createCheckpointDirectory(checkpointsDirectory, checkpointId);

            return new FsCheckpointStorageLocation(
                    fileSystem,
                    checkpointDir,
                    sharedStateDirectory,
                    taskOwnedStateDirectory,
                    reference,
                    fileSizeThreshold,
                    writeBufferSize);
        } else {
            // location encoded in the reference
            final Path path = decodePathFromReference(reference);

            return new FsCheckpointStorageLocation(
                    path.getFileSystem(),
                    path,
                    path,
                    path,
                    reference,
                    fileSizeThreshold,
                    writeBufferSize);
        }
    }
@Test
    void testResolveCheckpointStorageLocation() throws Exception {
        final FileSystem checkpointFileSystem = mock(FileSystem.class);
        final FsCheckpointStorageAccess storage =
                new FsCheckpointStorageAccess(
                        new TestingPath("hdfs:///checkpoint/", checkpointFileSystem),
                        null,
                        true,
                        new JobID(),
                        FILE_SIZE_THRESHOLD,
                        WRITE_BUFFER_SIZE);

        final FsCheckpointStorageLocation checkpointStreamFactory =
                (FsCheckpointStorageLocation)
                        storage.resolveCheckpointStorageLocation(
                                1L, CheckpointStorageLocationReference.getDefault());
        assertThat(checkpointStreamFactory.getFileSystem()).isEqualTo(checkpointFileSystem);

        final CheckpointStorageLocationReference savepointLocationReference =
                AbstractFsCheckpointStorageAccess.encodePathAsReference(
                        new Path("file:///savepoint/"));

        final FsCheckpointStorageLocation savepointStreamFactory =
                (FsCheckpointStorageLocation)
                        storage.resolveCheckpointStorageLocation(2L, savepointLocationReference);
        final FileSystem fileSystem = savepointStreamFactory.getFileSystem();
        assertThat(fileSystem).isInstanceOf(LocalFileSystem.class);
    }