private void registerStartingOffsets() {

    for (Map.Entry<String, SystemStream> changelogSystemStreamEntry : storeChangelogs.entrySet()) {
      SystemStreamPartition systemStreamPartition =
          new SystemStreamPartition(changelogSystemStreamEntry.getValue(), taskModel.getChangelogPartition());
      SystemAdmin systemAdmin = systemAdmins.getSystemAdmin(changelogSystemStreamEntry.getValue().getSystem());
      SystemConsumer systemConsumer = storeConsumers.get(changelogSystemStreamEntry.getKey());

      String offset = getStartingOffset(systemStreamPartition, systemAdmin);

      if (offset != null) {
        LOG.info("Registering change log consumer with offset " + offset + " for %" + systemStreamPartition);
        systemConsumer.register(systemStreamPartition, offset);
      } else {
        LOG.info("Skipping change log restoration for {} because stream appears to be empty (offset was null).",
            systemStreamPartition);
        taskStoresToRestore.remove(changelogSystemStreamEntry.getKey());
      }
    }
  }
Determines the starting offset for each store SSP (based on {@link #getStartingOffset(SystemStreamPartition, SystemAdmin)}) and
registers it with the respective SystemConsumer for starting consumption.
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java