private static Void cleanUpResources(Checkpoint checkpoint, Throwable throwable, long startTime,
      SamzaContainerMetrics samzaContainerMetrics, TaskName taskName, TaskRestoreManager taskRestoreManager,
      Map<TaskName, Checkpoint> newTaskCheckpoints) {
    // exception or not, update the restore time and close TaskRestoreManager as there will be no more retries.
    updateRestoreTime(startTime, samzaContainerMetrics, taskName);
    try {
      taskRestoreManager.close();
    } catch (Exception exception) {
      LOG.error("Error closing restore manager for task: {} after {} restore", taskName, exception);
      // ignore exception from close. container may still be able to continue processing/backups
      // if restore manager close fails.
    }
    if (throwable != null) {
      String msg = String.format("Error restoring state for task: %s", taskName.getTaskName());
      LOG.error(msg);
      throw new SamzaException(msg, throwable);
    }
    if (checkpoint != null) {
      newTaskCheckpoints.put(taskName, checkpoint);
    }
    return null;
  }
cleanup all the resources (like taskRestoreManager), update restore time and update task checkpoint
returned from the previous future in the chain
NOTE: This method should be called concurrently for all {@link TaskRestoreManager}s so that stop() can be parallelized.
      Certain persistent stores opened in BulkLoad mode are compacted on stop, so paralleling stop()
      also parallelizes their compaction (a time-intensive operation).
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerRestoreUtil.java