private CompletableFuture<Void> removeTTL(DirIndex dirIndex, Metadata metadata) {
    String dirName = dirIndex.getDirName();
    if (DirIndex.ROOT_DIR_NAME.equals(dirName)) {
      LOG.debug("Removing TTL for files and dirs present in DirIndex for root dir.");
    } else {
      LOG.debug("Removing TTL for files and dirs present in DirIndex for dir: {}", dirName);
    }

    List<CompletableFuture<Void>> updateTTLsFuture = new ArrayList<>();
    for (DirIndex subDir: dirIndex.getSubDirsPresent()) {
      updateTTLsFuture.add(removeTTL(subDir, metadata));
    }

    for (FileIndex file: dirIndex.getFilesPresent()) {
      Metadata requestMetadata =
          new Metadata(file.getFileName(), Optional.of(file.getFileMetadata().getSize()),
              metadata.getJobName(), metadata.getJobId(), metadata.getTaskName(), metadata.getStoreName());
      List<FileBlob> fileBlobs = file.getBlobs();
      for (FileBlob fileBlob : fileBlobs) {
        String opname = "removeTTL for fileBlob: " + file.getFileName() + " with blobId: {}" + fileBlob.getBlobId();
        Supplier<CompletionStage<Void>> ttlRemovalAction = () ->
            blobStoreManager.removeTTL(fileBlob.getBlobId(), requestMetadata).toCompletableFuture();
        CompletableFuture<Void> ttlRemovalFuture =
            FutureUtil.executeAsyncWithRetries(opname, ttlRemovalAction, isCauseNonRetriable(), executor, retryPolicyConfig);
        updateTTLsFuture.add(ttlRemovalFuture);
      }
    }

    return CompletableFuture.allOf(updateTTLsFuture.toArray(new CompletableFuture[0]));
  }
Recursively mark all the blobs associated with the {@link DirIndex} to never expire (remove TTL).
@param dirIndex the {@link DirIndex} whose contents' TTL needs to be removed
@param metadata {@link Metadata} related to the request
@return A future that completes when all the blobs associated with this dirIndex are marked to
never expire.
samza-core/src/main/java/org/apache/samza/storage/blobstore/util/BlobStoreUtil.java