@VisibleForTesting
  CompletableFuture<Void> getFile(List<FileBlob> fileBlobs, File fileToRestore, Metadata requestMetadata, boolean getDeleted) {
    FileOutputStream outputStream = null;
    try {
      long restoreFileStartTime = System.nanoTime();
      if (fileToRestore.exists()) {
        // delete the file if it already exists, e.g. from a previous retry.
        Files.delete(fileToRestore.toPath());
      }

      outputStream = new FileOutputStream(fileToRestore);
      final FileOutputStream finalOutputStream = outputStream;
      // TODO HIGH shesharm add integration tests to ensure empty files and directories are handled correctly E2E.
      fileToRestore.createNewFile(); // create file for 0 byte files (fileIndex entry but no fileBlobs).
      // create a copy to ensure list being sorted is mutable.
      List<FileBlob> fileBlobsCopy = new ArrayList<>(fileBlobs);
      fileBlobsCopy.sort(Comparator.comparingInt(FileBlob::getOffset)); // sort by offset.

      // chain the futures such that write to file for blobs is sequential.
      // can be optimized to write concurrently to the file later.
      CompletableFuture<Void> resultFuture = CompletableFuture.completedFuture(null);
      for (FileBlob fileBlob : fileBlobsCopy) {
        resultFuture = resultFuture.thenComposeAsync(v -> {
          LOG.debug("Starting restore for file: {} with blob id: {} at offset: {} with getDeleted set to: {}",
              fileToRestore, fileBlob.getBlobId(), fileBlob.getOffset(), getDeleted);
          return blobStoreManager.get(fileBlob.getBlobId(), finalOutputStream, requestMetadata, getDeleted);
        }, executor);
      }

      resultFuture = resultFuture.thenRunAsync(() -> {
        LOG.debug("Finished restore for file: {}. Closing output stream.", fileToRestore);
        try {
          // flush the file contents to disk
          finalOutputStream.getFD().sync();
          finalOutputStream.close();
        } catch (Exception e) {
          throw new SamzaException(String.format("Error closing output stream for file: %s", fileToRestore.getAbsolutePath()), e);
        }
      }, executor);

      resultFuture.whenComplete((res, ex) -> {
        if (restoreMetrics != null) {
          restoreMetrics.avgFileRestoreNs.update(System.nanoTime() - restoreFileStartTime);

          long fileSize = requestMetadata.getPayloadSize();
          restoreMetrics.restoreRate.inc(fileSize);
          restoreMetrics.filesRestored.getValue().addAndGet(1);
          restoreMetrics.bytesRestored.getValue().addAndGet(fileSize);
          restoreMetrics.filesRemaining.getValue().addAndGet(-1);
          restoreMetrics.bytesRemaining.getValue().addAndGet(-1 * fileSize);
        }
      });
      return resultFuture;
    } catch (Exception exception) {
      try {
        if (outputStream != null) {
          outputStream.close();
        }
      } catch (Exception err) {
        LOG.error("Error closing output stream for file: {}", fileToRestore.getAbsolutePath(), err);
      }

      throw new SamzaException(String.format("Error restoring file: %s in path: %s",
          fileToRestore.getName(), requestMetadata.getPayloadPath()), exception);
    }
  }
Gets a file from the blob store.
@param fileBlobs List of {@link FileBlob}s that constitute this file.
@param fileToRestore File pointing to the local path where the file will be restored.
@param requestMetadata {@link Metadata} associated with this request
@param getDeleted Flag that indicates whether to try to get Deleted (but not yet compacted) files.
@return a future that completes when the file is downloaded and written or if an exception occurs.
samza-core/src/main/java/org/apache/samza/storage/blobstore/util/BlobStoreUtil.java