public void transferAllStateDataToDirectory(
            Collection<StateHandleDownloadSpec> downloadRequests,
            CloseableRegistry closeableRegistry)
            throws Exception {

        // We use this closer for fine-grained shutdown of all parallel downloading.
        CloseableRegistry internalCloser = new CloseableRegistry();
        // Make sure we also react to external close signals.
        closeableRegistry.registerCloseable(internalCloser);
        try {
            // We have to wait for all futures to be completed, to make sure in
            // case of failure that we will clean up all the files
            FutureUtils.completeAll(
                            createDownloadRunnables(downloadRequests, internalCloser).stream()
                                    .map(
                                            runnable ->
                                                    CompletableFuture.runAsync(
                                                            runnable,
                                                            transfer.getExecutorService()))
                                    .collect(Collectors.toList()))
                    .get();
        } catch (Exception e) {
            downloadRequests.stream()
                    .map(StateHandleDownloadSpec::getDownloadDestination)
                    .map(Path::toFile)
                    .forEach(FileUtils::deleteDirectoryQuietly);
            // Error reporting
            Throwable throwable = ExceptionUtils.stripExecutionException(e);
            throwable = ExceptionUtils.stripException(throwable, RuntimeException.class);
            if (throwable instanceof IOException) {
                throw (IOException) throwable;
            } else {
                throw new FlinkRuntimeException("Failed to download data for state handles.", e);
            }
        } finally {
            // Unregister and close the internal closer.
            if (closeableRegistry.unregisterCloseable(internalCloser)) {
                IOUtils.closeQuietly(internalCloser);
            }
        }
    }
@Test
    public void testMultiThreadRestoreThreadPoolExceptionRethrow() {
        SpecifiedException expectedCause =
                new SpecifiedException("throw exception while multi thread restore.");
        StreamStateHandle stateHandle = new ThrowingStateHandle(expectedCause);

        List<HandleAndLocalPath> stateHandles = new ArrayList<>(1);
        stateHandles.add(HandleAndLocalPath.of(stateHandle, "state1"));

        IncrementalRemoteKeyedStateHandle incrementalKeyedStateHandle =
                new IncrementalRemoteKeyedStateHandle(
                        UUID.randomUUID(),
                        KeyGroupRange.EMPTY_KEY_GROUP_RANGE,
                        1,
                        stateHandles,
                        stateHandles,
                        stateHandle);

        try (RocksDBStateDownloader rocksDBStateDownloader = new RocksDBStateDownloader(5)) {
            rocksDBStateDownloader.transferAllStateDataToDirectory(
                    Collections.singletonList(
                            new StateHandleDownloadSpec(
                                    incrementalKeyedStateHandle,
                                    temporaryFolder.newFolder().toPath())),
                    new CloseableRegistry());
            fail();
        } catch (Exception e) {
            assertEquals(expectedCause, e.getCause());
        }
    }