private Map<String, KafkaStateCheckpointMarker> getCheckpointedChangelogOffsets(Checkpoint checkpoint) {
    Map<String, KafkaStateCheckpointMarker> checkpointedChangelogOffsets = new HashMap<>();
    if (checkpoint == null) return checkpointedChangelogOffsets;

    if (checkpoint instanceof CheckpointV2) {
      Map<String, Map<String, String>> factoryStoreSCMs = ((CheckpointV2) checkpoint).getStateCheckpointMarkers();
      if (factoryStoreSCMs.containsKey(KafkaStateCheckpointMarker.KAFKA_STATE_BACKEND_FACTORY_NAME)) {
        factoryStoreSCMs.get(KafkaStateCheckpointMarker.KAFKA_STATE_BACKEND_FACTORY_NAME)
            .forEach((storeName, scmString) -> {
              KafkaStateCheckpointMarker kafkaSCM = KafkaStateCheckpointMarker.deserialize(scmString);
              checkpointedChangelogOffsets.put(storeName, kafkaSCM);
            });
      } // skip the non-KafkaStateCheckpointMarkers
    } else if (checkpoint instanceof CheckpointV1) {
      // If the checkpoint v1 is used, we need to fetch the changelog SSPs in the inputOffsets in order to get the
      // store offset.
      Map<SystemStreamPartition, String> checkpointedOffsets = checkpoint.getOffsets();
      storeChangelogs.forEach((storeName, systemStream) -> {
        Partition changelogPartition = taskModel.getChangelogPartition();
        SystemStreamPartition storeChangelogSSP = new SystemStreamPartition(systemStream, changelogPartition);
        String checkpointedOffset = checkpointedOffsets.get(storeChangelogSSP);
        if (StringUtils.isNotBlank(checkpointedOffset)) {
          KafkaChangelogSSPOffset kafkaChangelogSSPOffset = KafkaChangelogSSPOffset.fromString(checkpointedOffset);
          KafkaStateCheckpointMarker marker = new KafkaStateCheckpointMarker(
              storeChangelogSSP, kafkaChangelogSSPOffset.getChangelogOffset());
          checkpointedChangelogOffsets.put(storeName, marker);
        }
      });
    } else {
      throw new SamzaException("Unsupported checkpoint version: " + checkpoint.getVersion());
    }

    return checkpointedChangelogOffsets;
  }
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