private boolean isLoggedStoreValid(String storeName, File loggedStoreDir) {
    long changeLogDeleteRetentionInMs = new StorageConfig(config).getChangeLogDeleteRetentionInMs(storeName);

    if (storeChangelogs.containsKey(storeName)) {
      SystemStreamPartition changelogSSP = new SystemStreamPartition(storeChangelogs.get(storeName), taskModel.getChangelogPartition());
      return this.taskStores.get(storeName).getStoreProperties().isPersistedToDisk()
          && storageManagerUtil.isOffsetFileValid(loggedStoreDir, Collections.singleton(changelogSSP), false)
          && !storageManagerUtil.isStaleStore(loggedStoreDir, changeLogDeleteRetentionInMs, clock.currentTimeMillis(), false);
    }

    return false;
  }
Directory loggedStoreDir associated with the logged store storeName is determined to be valid
if all of the following conditions are true.
a) If the store has to be persisted to disk.
b) If there is a valid offset file associated with the logged store.
c) If the logged store has not gone stale.

@return true if the logged store is valid, false otherwise.
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java