@VisibleForTesting
  static boolean shouldRestore(String taskName, String storeName, DirIndex dirIndex,
      Path storeCheckpointDir, StorageConfig storageConfig, DirDiffUtil dirDiffUtil) {
    // if a store checkpoint directory exists for the last successful task checkpoint, try to use it.
    boolean restoreStore;
    if (Files.exists(storeCheckpointDir)) {
      if (storageConfig.cleanLoggedStoreDirsOnStart(storeName)) {
        LOG.debug("Restoring task: {} store: {} from remote snapshot since the store is configured to be " +
            "restored on each restart.", taskName, storeName);
        restoreStore = true;
      } else if (dirDiffUtil.areSameDir(FILES_TO_IGNORE, false, true).test(storeCheckpointDir.toFile(), dirIndex)) {
        restoreStore = false; // no restore required for this store.
      } else {
        // we don't optimize for the case when the local host doesn't contain the most recent store checkpoint
        // directory but contains an older checkpoint directory which could have partial overlap with the remote
        // snapshot. we also don't try to optimize for any edge cases where the most recent checkpoint directory
        // contents could be partially different than the remote store (afaik, there is no known valid scenario
        // where this could happen right now, except for the offset file handling above).
        // it's simpler and fast enough for now to restore the entire store instead.

        LOG.error("Local store checkpoint directory: {} contents are not the same as the remote snapshot. " +
            "Queuing for restore from remote snapshot.", storeCheckpointDir);
        restoreStore = true;
      }
    } else { // did not find last checkpoint dir, restore the store from the remote blob store
      LOG.debug("No local store checkpoint directory found at: {}. " +
          "Queuing for restore from remote snapshot.", storeCheckpointDir);
      restoreStore = true;
    }

    return restoreStore;
  }
Determines if the store needs to be restored from remote snapshot based on local and remote state.
samza-core/src/main/java/org/apache/samza/storage/blobstore/BlobStoreRestoreManager.java