public List<HandleAndLocalPath> uploadFilesToCheckpointFs(
            @Nonnull List<Path> files,
            CheckpointStreamFactory checkpointStreamFactory,
            CheckpointedStateScope stateScope,
            CloseableRegistry closeableRegistry,
            CloseableRegistry tmpResourcesRegistry)
            throws Exception {

        List<CompletableFuture<HandleAndLocalPath>> futures =
                createUploadFutures(
                        files,
                        checkpointStreamFactory,
                        stateScope,
                        closeableRegistry,
                        tmpResourcesRegistry);

        List<HandleAndLocalPath> handles = new ArrayList<>(files.size());

        try {
            FutureUtils.waitForAll(futures).get();

            for (CompletableFuture<HandleAndLocalPath> future : futures) {
                handles.add(future.get());
            }
        } catch (ExecutionException e) {
            Throwable throwable = ExceptionUtils.stripExecutionException(e);
            throwable = ExceptionUtils.stripException(throwable, RuntimeException.class);
            if (throwable instanceof IOException) {
                throw (IOException) throwable;
            } else {
                throw new FlinkRuntimeException("Failed to upload data for state handles.", e);
            }
        }

        return handles;
    }
@Test
    void testMultiThreadUploadCorrectly() throws Exception {
        File checkpointPrivateFolder = TempDirUtils.newFolder(temporaryFolder, "private");
        org.apache.flink.core.fs.Path checkpointPrivateDirectory =
                org.apache.flink.core.fs.Path.fromLocalFile(checkpointPrivateFolder);

        File checkpointSharedFolder = TempDirUtils.newFolder(temporaryFolder, "shared");
        org.apache.flink.core.fs.Path checkpointSharedDirectory =
                org.apache.flink.core.fs.Path.fromLocalFile(checkpointSharedFolder);

        FileSystem fileSystem = checkpointPrivateDirectory.getFileSystem();
        int fileStateSizeThreshold = 1024;
        int writeBufferSize = 4096;
        FsCheckpointStreamFactory checkpointStreamFactory =
                new FsCheckpointStreamFactory(
                        fileSystem,
                        checkpointPrivateDirectory,
                        checkpointSharedDirectory,
                        fileStateSizeThreshold,
                        writeBufferSize);

        String localFolder = "local";
        TempDirUtils.newFolder(temporaryFolder, localFolder);

        int sstFileCount = 6;
        List<Path> sstFilePaths =
                generateRandomSstFiles(localFolder, sstFileCount, fileStateSizeThreshold);

        try (RocksDBStateUploader rocksDBStateUploader = new RocksDBStateUploader(5)) {
            List<HandleAndLocalPath> sstFiles =
                    rocksDBStateUploader.uploadFilesToCheckpointFs(
                            sstFilePaths,
                            checkpointStreamFactory,
                            CheckpointedStateScope.SHARED,
                            new CloseableRegistry(),
                            new CloseableRegistry());

            for (Path path : sstFilePaths) {
                assertStateContentEqual(
                        path,
                        sstFiles.stream()
                                .filter(e -> e.getLocalPath().equals(path.getFileName().toString()))
                                .findFirst()
                                .get()
                                .getHandle()
                                .openInputStream());
            }
        }
    }