public Map<SystemStreamPartition, String> readOffsetFile(File storagePartitionDir, Set<SystemStreamPartition> storeSSPs, boolean isSideInput) {

    File offsetFileRefNew = new File(storagePartitionDir, OFFSET_FILE_NAME_NEW);
    File offsetFileRefLegacy = new File(storagePartitionDir, OFFSET_FILE_NAME_LEGACY);

    // First we check if the new offset file exists, if it does we read offsets from it regardless of old or new format,
    // if it doesn't exist, we check if the store is non-sideInput and legacy-offset file exists, if so we read offsets
    // from the old non-side-input offset file (regardless of the offset format),
    // last, we check if the store is a sideInput and the old side-input-offset file exists
    if (offsetFileRefNew.exists()) {
      return readOffsetFile(storagePartitionDir, offsetFileRefNew.getName(), storeSSPs);
    } else if (!isSideInput && offsetFileRefLegacy.exists()) {
      return readOffsetFile(storagePartitionDir, offsetFileRefLegacy.getName(), storeSSPs);
    } else if (isSideInput) {
      Map<SystemStreamPartition, String> result = new HashMap<>();
      SIDE_INPUT_OFFSET_FILE_SEMAPHORE.acquireUninterruptibly();
      try {
        File sideInputOffsetFileRefLegacy = new File(storagePartitionDir, SIDE_INPUT_OFFSET_FILE_NAME_LEGACY);
        if (sideInputOffsetFileRefLegacy.exists()) {
          result = readOffsetFile(storagePartitionDir, sideInputOffsetFileRefLegacy.getName(), storeSSPs);
        }
      } finally {
        SIDE_INPUT_OFFSET_FILE_SEMAPHORE.release();
      }
      return result;
    } else {
      return new HashMap<>();
    }
  }
Read and return the offset from the directory's offset file

@param storagePartitionDir the base directory of the store
@param storeSSPs SSPs associated with the store (if any)
@param isSideInput, true if the store is a side-input store, false otherwise
@return the content of the offset file if it exists for the store, null otherwise.
samza-core/src/main/java/org/apache/samza/storage/StorageManagerUtil.java