private CheckpointId getCheckpointId(Checkpoint checkpoint) {
    if (checkpoint == null) return null;
    if (checkpoint instanceof CheckpointV1) {
      for (Map.Entry<String, SystemStream> storeNameSystemStream : storeChangelogs.entrySet()) {
        SystemStreamPartition storeChangelogSSP = new SystemStreamPartition(storeNameSystemStream.getValue(), taskModel.getChangelogPartition());
        String checkpointMessage = checkpoint.getOffsets().get(storeChangelogSSP);
        if (StringUtils.isNotBlank(checkpointMessage)) {
          KafkaChangelogSSPOffset kafkaStateChanglogOffset = KafkaChangelogSSPOffset.fromString(checkpointMessage);
          return kafkaStateChanglogOffset.getCheckpointId();
        }
      }
    } else if (checkpoint instanceof CheckpointV2) {
      return ((CheckpointV2) checkpoint).getCheckpointId();
    } else {
      throw new SamzaException("Unsupported checkpoint version: " + checkpoint.getVersion());
    }
    return null;
  }
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