public CompletionStage<DirIndex> putDir(DirDiff dirDiff, SnapshotMetadata snapshotMetadata) {
    // Upload all new files in the dir
    List<File> filesToUpload = dirDiff.getFilesAdded();
    List<CompletionStage<FileIndex>> fileFutures = filesToUpload.stream()
        .map(file -> putFile(file, snapshotMetadata))
        .collect(Collectors.toList());

    CompletableFuture<Void> allFilesFuture =
        CompletableFuture.allOf(fileFutures.toArray(new CompletableFuture[0]));

    List<CompletionStage<DirIndex>> subDirFutures = new ArrayList<>();
    // recursively upload all new subdirs of this dir
    for (DirDiff subDirAdded: dirDiff.getSubDirsAdded()) {
      subDirFutures.add(putDir(subDirAdded, snapshotMetadata));
    }
    // recursively update contents of all subdirs that are retained but might have been modified
    for (DirDiff subDirRetained: dirDiff.getSubDirsRetained()) {
      subDirFutures.add(putDir(subDirRetained, snapshotMetadata));
    }
    CompletableFuture<Void> allDirBlobsFuture =
        CompletableFuture.allOf(subDirFutures.toArray(new CompletableFuture[0]));

    return CompletableFuture.allOf(allDirBlobsFuture, allFilesFuture)
        .thenApplyAsync(f -> {
          LOG.trace("All file and dir uploads complete for task: {} store: {}",
              snapshotMetadata.getTaskName(), snapshotMetadata.getStoreName());
          List<FileIndex> filesPresent = fileFutures.stream()
              .map(blob -> blob.toCompletableFuture().join())
              .collect(Collectors.toList());

          filesPresent.addAll(dirDiff.getFilesRetained());

          List<DirIndex> subDirsPresent = subDirFutures.stream()
              .map(subDir -> subDir.toCompletableFuture().join())
              .collect(Collectors.toList());

          LOG.debug("Uploaded diff for task: {} store: {} with statistics: {}",
              snapshotMetadata.getTaskName(), snapshotMetadata.getStoreName(), DirDiff.getStats(dirDiff));

          LOG.trace("Returning new DirIndex for task: {} store: {}",
              snapshotMetadata.getTaskName(), snapshotMetadata.getStoreName());
          return new DirIndex(dirDiff.getDirName(),
              filesPresent,
              dirDiff.getFilesRemoved(),
              subDirsPresent,
              dirDiff.getSubDirsRemoved());
        }, executor);
  }
Recursively upload all new files and upload or update contents of all subdirs in the {@link DirDiff} and return a
Future containing the {@link DirIndex} associated with the directory.
@param dirDiff diff for the contents of this directory
@return A future with the {@link DirIndex} if the upload completed successfully.
samza-core/src/main/java/org/apache/samza/storage/blobstore/util/BlobStoreUtil.java