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 testMultiThreadUploadThreadPoolExceptionRethrow() throws IOException {
        SpecifiedException expectedException =
                new SpecifiedException("throw exception while multi thread upload states.");

        CheckpointStateOutputStream outputStream =
                createFailingCheckpointStateOutputStream(expectedException);
        CheckpointStreamFactory checkpointStreamFactory =
                new CheckpointStreamFactory() {
                    @Override
                    public CheckpointStateOutputStream createCheckpointStateOutputStream(
                            CheckpointedStateScope scope) {
                        return outputStream;
                    }

                    @Override
                    public boolean canFastDuplicate(
                            StreamStateHandle stateHandle, CheckpointedStateScope scope) {
                        return false;
                    }

                    @Override
                    public List<StreamStateHandle> duplicate(
                            List<StreamStateHandle> stateHandles, CheckpointedStateScope scope) {
                        return null;
                    }
                };

        File file = TempDirUtils.newFile(temporaryFolder, String.valueOf(UUID.randomUUID()));
        generateRandomFileContent(file.getPath(), 20);

        List<Path> filePaths = new ArrayList<>(1);
        filePaths.add(file.toPath());
        try (RocksDBStateUploader rocksDBStateUploader = new RocksDBStateUploader(5)) {
            assertThatThrownBy(
                            () ->
                                    rocksDBStateUploader.uploadFilesToCheckpointFs(
                                            filePaths,
                                            checkpointStreamFactory,
                                            CheckpointedStateScope.SHARED,
                                            new CloseableRegistry(),
                                            new CloseableRegistry()))
                    .isEqualTo(expectedException);
        }
    }