private static void updateRestoreTime(long startTime, SamzaContainerMetrics samzaContainerMetrics,
      TaskName taskInstance) {
    long timeToRestore = System.currentTimeMillis() - startTime;
    if (samzaContainerMetrics != null) {
      Gauge taskGauge = samzaContainerMetrics.taskStoreRestorationMetrics().getOrDefault(taskInstance, null);

      if (taskGauge != null) {
        taskGauge.set(timeToRestore);
      }
    }
  }
Returns a single future that guarantees all the following are completed, in this order:
  1. Restore state locally by getting deleted blobs from the blob store.
  2. Create a new snapshot from restored state by backing it up on the blob store.
  3. Remove TTL from the new Snapshot and all the associated blobs in the blob store
  4. Clean up old/deleted Snapshot
  5. Create and write the new checkpoint to checkpoint topic
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerRestoreUtil.java