public boolean isStaleStore(File storeDir, long storeDeleteRetentionInMs, long currentTimeMs, boolean isSideInput) {
    long offsetFileLastModifiedTime;
    boolean isStaleStore = false;
    String storePath = storeDir.toPath().toString();

    if (storeDir.exists()) {

      // We check if the new offset-file exists, if so we use its last-modified time, if it doesn't we use the legacy file
      // depending on if it is a side-input or not,
      // if neither exists, we use 0L (the default return value of lastModified() when file does not exist
      File offsetFileRefNew = new File(storeDir, OFFSET_FILE_NAME_NEW);
      File offsetFileRefLegacy = new File(storeDir, OFFSET_FILE_NAME_LEGACY);
      File checkpointV2File = new File(storeDir, CHECKPOINT_FILE_NAME);

      if (checkpointV2File.exists()) {
        offsetFileLastModifiedTime = checkpointV2File.lastModified();
      } else if (offsetFileRefNew.exists()) {
        offsetFileLastModifiedTime = offsetFileRefNew.lastModified();
      } else if (!isSideInput && offsetFileRefLegacy.exists()) {
        offsetFileLastModifiedTime = offsetFileRefLegacy.lastModified();
      } else if (isSideInput) {
        try {
          SIDE_INPUT_OFFSET_FILE_SEMAPHORE.acquireUninterruptibly();
          File sideInputOffsetFileRefLegacy = new File(storeDir, SIDE_INPUT_OFFSET_FILE_NAME_LEGACY);
          if (sideInputOffsetFileRefLegacy.exists()) {
            offsetFileLastModifiedTime = sideInputOffsetFileRefLegacy.lastModified();
          } else {
            offsetFileLastModifiedTime = 0L;
          }
        } finally {
          SIDE_INPUT_OFFSET_FILE_SEMAPHORE.release();
        }
      } else {
        offsetFileLastModifiedTime = 0L;
      }

      if ((currentTimeMs - offsetFileLastModifiedTime) >= storeDeleteRetentionInMs) {
        LOG.info(
            String.format("Store: %s is stale since lastModifiedTime of offset file: %d, is older than store deleteRetentionMs: %d.",
            storePath, offsetFileLastModifiedTime, storeDeleteRetentionInMs));
        isStaleStore = true;
      }
    } else {
      LOG.info("Storage partition directory: {} does not exist.", storePath);
    }
    return isStaleStore;
  }
Checks if the store is stale. If the time elapsed since the last modified time of the offset file is greater than
the {@code storeDeleteRetentionInMs}, then the store is considered stale.

@param storeDir the base directory of the store
@param storeDeleteRetentionInMs store delete retention in millis
@param currentTimeMs current time in ms
@param isSideInput true if store is a side-input store, false if it is a regular store
@return true if the store is stale, false otherwise
samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java