@VisibleForTesting
  static Map<SystemStreamPartition, SystemStreamPartitionMetadata> getCurrentChangelogOffsets(
      TaskModel taskModel, Map<String, SystemStream> storeChangelogs, SSPMetadataCache sspMetadataCache) {
    Map<SystemStreamPartition, SystemStreamPartitionMetadata> changelogOffsets = new HashMap<>();

    Partition changelogPartition = taskModel.getChangelogPartition();
    for (Map.Entry<String, SystemStream> storeChangelog : storeChangelogs.entrySet()) {
      SystemStream changelog = storeChangelog.getValue();
      SystemStreamPartition changelogSSP = new SystemStreamPartition(
          changelog.getSystem(), changelog.getStream(), changelogPartition);
      SystemStreamPartitionMetadata metadata = sspMetadataCache.getMetadata(changelogSSP);
      changelogOffsets.put(changelogSSP, metadata);
    }

    LOG.info("Got current changelog offsets for taskName: {} as: {}", taskModel.getTaskName(), changelogOffsets);
    return changelogOffsets;
  }
Get offset metadata for each changelog SSP for this task. A task may have multiple changelog streams
(e.g., for different stores), but will have the same partition for all of them.
samza-core/src/main/java/org/apache/samza/storage/TransactionalStateTaskRestoreManager.java