public void init(Checkpoint checkpoint, boolean getDeleted) {
    long startTime = System.nanoTime();
    LOG.debug("Initializing blob store restore manager for task {} with getDeleted {}", taskName, getDeleted);

    blobStoreManager.init();

    // get previous SCMs from checkpoint
    prevStoreSnapshotIndexes =
        blobStoreUtil.getStoreSnapshotIndexes(jobName, jobId, taskName, checkpoint, storesToRestore, getDeleted);
    metrics.getSnapshotIndexNs.set(System.nanoTime() - startTime);
    LOG.trace("Found previous snapshot index during blob store restore manager init for task: {} to be: {} with getDeleted set to {}",
        taskName, prevStoreSnapshotIndexes, getDeleted);

    metrics.initStoreMetrics(storesToRestore);

    // Note: blocks the caller thread.
    deleteUnusedStoresFromBlobStore(jobName, jobId, taskName, storageConfig, blobStoreConfig, prevStoreSnapshotIndexes,
        blobStoreUtil, executor);
    metrics.initNs.set(System.nanoTime() - startTime);
  }
Initialize state resources such as store directories.
NOTE: init can be called twice. In case init fails with DeletedException for the first time, it will be retried with
      getDeleted set to true.
@param checkpoint Current task checkpoint
@param getDeleted Flag to get deleted SnapshotIndex blob
samza-core/src/main/java/org/apache/samza/storage/blobstore/BlobStoreRestoreManager.java