code: public Map<StateHandleID, StreamStateHandle> uploadFilesToCheckpointFs(
  		@Nonnull Map<StateHandleID, Path> files,
  		CheckpointStreamFactory checkpointStreamFactory,
  		CloseableRegistry closeableRegistry) throws Exception {
  
  		Map<StateHandleID, StreamStateHandle> handles = new HashMap<>();
  
  		Map<StateHandleID, CompletableFuture<StreamStateHandle>> futures =
  			createUploadFutures(files, checkpointStreamFactory, closeableRegistry);
  
  		try {
  			FutureUtils.waitForAll(futures.values()).get();
  
  			for (Map.Entry<StateHandleID, CompletableFuture<StreamStateHandle>> entry : futures.entrySet()) {
  				handles.put(entry.getKey(), entry.getValue().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 download data for state handles.", e);
  			}
  		}
  
  		return handles;
  	}
docs: Upload all the files to checkpoint fileSystem using specified number of threads.
  
  @param files The files will be uploaded to checkpoint filesystem.
  @param checkpointStreamFactory The checkpoint streamFactory used to create outputstream.
  
  @throws Exception Thrown if can not upload all the files.