@VisibleForTesting
  static void registerStartingOffsets(
      TaskModel taskModel,
      StoreActions storeActions,
      Map<String, SystemStream> storeChangelogs,
      SystemAdmins systemAdmins,
      Map<String, SystemConsumer> storeConsumers,
      Map<SystemStreamPartition, SystemStreamPartitionMetadata> currentChangelogOffsets) {
    Map<String, RestoreOffsets> storesToRestore = storeActions.storesToRestore;

    // must register at least one SSP with each changelog system consumer otherwise start will throw.
    // hence we register upcoming offset as the dummy offset by default and override it later if necessary.
    // using upcoming offset ensures that no messages are replayed by default.
    storeChangelogs.forEach((storeName, changelog) -> {
      SystemStreamPartition changelogSSP = new SystemStreamPartition(changelog, taskModel.getChangelogPartition());
      SystemConsumer systemConsumer = storeConsumers.get(storeName);
      SystemStreamPartitionMetadata currentOffsets = currentChangelogOffsets.get(changelogSSP);
      String upcomingOffset = currentOffsets.getUpcomingOffset();
      LOG.info("Temporarily registering upcoming offset: {} as the starting offest for changelog ssp: {}. " +
          "This might be overridden later for stores that need restoring.", upcomingOffset, changelogSSP);
      systemConsumer.register(changelogSSP, upcomingOffset);
    });

    // now register the actual starting offset if necessary. system consumer will ensure that the lower of the
    // two registered offsets is used as the starting offset.
    storesToRestore.forEach((storeName, restoreOffsets) -> {
      SystemStream changelog = storeChangelogs.get(storeName);
      SystemStreamPartition changelogSSP = new SystemStreamPartition(changelog, taskModel.getChangelogPartition());
      SystemAdmin systemAdmin = systemAdmins.getSystemAdmin(changelog.getSystem());
      validateRestoreOffsets(restoreOffsets, systemAdmin);

      SystemConsumer systemConsumer = storeConsumers.get(storeName);
      SystemStreamPartitionMetadata currentOffsets = currentChangelogOffsets.get(changelogSSP);
      String oldestOffset = currentOffsets.getOldestOffset();

      // if the starting offset equals oldest offset (e.g. for full restore), start from the oldest offset (inclusive).
      // else, start from the next (upcoming) offset.
      String startingOffset;
      if (systemAdmin.offsetComparator(restoreOffsets.startingOffset, oldestOffset) == 0) {
        startingOffset = oldestOffset;
      } else {
        Map<SystemStreamPartition, String> offsetMap = ImmutableMap.of(changelogSSP, restoreOffsets.startingOffset);
        startingOffset = systemAdmin.getOffsetsAfter(offsetMap).get(changelogSSP);
      }
      LOG.info("Registering starting offset: {} for changelog ssp: {}", startingOffset, changelogSSP);
      systemConsumer.register(changelogSSP, startingOffset);
    });
  }
Determines the starting offset for each store changelog SSP that needs to be restored from,
and registers it with the respective SystemConsumer.
samza-core/src/main/java/org/apache/samza/storage/TransactionalStateTaskRestoreManager.java