public boolean isOffsetFileValid(File storeDir, Set<SystemStreamPartition> storeSSPs, boolean isSideInput) {
    boolean hasValidOffsetFile = false;
    if (storeDir.exists()) {
      Map<SystemStreamPartition, String> offsetContents = readOffsetFile(storeDir, storeSSPs, isSideInput);
      if (offsetContents == null) {
        LOG.info("Offset file is invalid since it does not exist. Store directory: {}", storeDir.toPath());
      } else if (offsetContents.isEmpty()) {
        LOG.info("Offset file is invalid since it is empty. Store directory: {}", storeDir.toPath());
      } else if (!offsetContents.keySet().equals(storeSSPs)) {
        LOG.info("Offset file is invalid since changelog or side input SSPs don't match. "
            + "Store directory: {}. SSPs from offset-file: {} SSPs expected: {} ",
            storeDir.toPath(), offsetContents.keySet(), storeSSPs);
      } else {
        hasValidOffsetFile = true;
      }
    }

    return hasValidOffsetFile;
  }
An offset file associated with logged store {@code storeDir} is valid if it exists, is not empty,
and the offsets are for the store's current changelog or side input SSPs.

@param storeDir the base directory of the store
@param storeSSPs ssps (if any) associated with the store
@param isSideInput true if store is a side-input store, false if it is a regular store
@return true if the offset file is valid. false otherwise.
samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java