private static void validateRestoreOffsets(RestoreOffsets restoreOffsets, SystemAdmin systemAdmin) {
    String startingOffset = restoreOffsets.startingOffset;
    String endingOffset = restoreOffsets.endingOffset;
    // cannot enforce that starting offset be non null since SSPMetadata allows oldest offset == null for empty topics.
    if (endingOffset != null) {
      Preconditions.checkState(systemAdmin.offsetComparator(endingOffset, startingOffset) >= 0,
          String.format("Ending offset: %s must be equal to or greater than starting offset: %s",
              endingOffset, 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