public CompletableFuture<Void> restoreDir(File baseDir, DirIndex dirIndex, Metadata metadata, boolean getDeleted) {
    LOG.debug("Restoring contents of directory: {} from remote snapshot. GetDeletedFiles set to: {}", baseDir, getDeleted);

    List<CompletableFuture<Void>> downloadFutures = new ArrayList<>();

    try {
      // create parent directories if they don't exist
      Files.createDirectories(baseDir.toPath());
    } catch (IOException exception) {
      LOG.error("Error creating directory: {} for restore", baseDir.getAbsolutePath(), exception);
      throw new SamzaException(String.format("Error creating directory: %s for restore",
          baseDir.getAbsolutePath()), exception);
    }

    // restore all files in the directory
    for (FileIndex fileIndex : dirIndex.getFilesPresent()) {
      File fileToRestore = Paths.get(baseDir.getAbsolutePath(), fileIndex.getFileName()).toFile();
      Metadata requestMetadata =
          new Metadata(fileToRestore.getAbsolutePath(), Optional.of(fileIndex.getFileMetadata().getSize()),
              metadata.getJobName(), metadata.getJobId(), metadata.getTaskName(), metadata.getStoreName());
      List<FileBlob> fileBlobs = fileIndex.getBlobs();

      String opName = "restoreFile: " + fileToRestore.getAbsolutePath();
      CompletableFuture<Void> fileRestoreFuture =
          FutureUtil.executeAsyncWithRetries(opName, () -> getFile(fileBlobs, fileToRestore, requestMetadata, getDeleted),
              isCauseNonRetriable(), executor, retryPolicyConfig);
      downloadFutures.add(fileRestoreFuture);
    }

    // restore any sub-directories
    List<DirIndex> subDirs = dirIndex.getSubDirsPresent();
    for (DirIndex subDir : subDirs) {
      File subDirFile = Paths.get(baseDir.getAbsolutePath(), subDir.getDirName()).toFile();
      downloadFutures.add(restoreDir(subDirFile, subDir, metadata, getDeleted));
    }

    return FutureUtil.allOf(downloadFutures);
  }
Non-blocking restore of a {@link SnapshotIndex} to local store by downloading all the files and sub-dirs associated
with this remote snapshot.
NOTE: getDeleted flag sets if it reattempts to get a deleted file by setting getDeleted flag in getFiles.
@return A future that completes when all the async downloads completes
samza-core/src/main/java/org/apache/samza/storage/blobstore/util/BlobStoreUtil.java