@VisibleForTesting
  Map<SystemStreamPartition, String> getStartingOffsets(
      Map<SystemStreamPartition, String> fileOffsets, Map<SystemStreamPartition, String> oldestOffsets) {
    Map<SystemStreamPartition, String> startingOffsets = new HashMap<>();

    this.sspToStores.keySet().forEach(ssp -> {
      String fileOffset = fileOffsets.get(ssp);
      String oldestOffset = oldestOffsets.get(ssp);

      startingOffsets.put(ssp,
          this.storageManagerUtil.getStartingOffset(
              ssp, this.systemAdmins.getSystemAdmin(ssp.getSystem()), fileOffset, oldestOffset));
    });

    return startingOffsets;
  }
Gets the starting offsets for the {@link SystemStreamPartition}s belonging to all the side input stores. See doc
of {@link StorageManagerUtil#getStartingOffset} for how file offsets and oldest offsets for each SSP are
reconciled.

@param fileOffsets offsets from the local offset file
@param oldestOffsets oldest offsets from the source
@return a {@link Map} of {@link SystemStreamPartition} to offset
samza-core/src/main/java/org/apache/samza/storage/TaskSideInputHandler.java