private void getOldestChangeLogOffsets() {

    Map<SystemStream, SystemStreamMetadata> changeLogMetadata = JavaConverters.mapAsJavaMapConverter(
        streamMetadataCache.getStreamMetadata(
            JavaConverters.asScalaSetConverter(new HashSet<>(storeChangelogs.values())).asScala().toSet(),
            false)).asJava();

    LOG.info("Got change log stream metadata: {}", changeLogMetadata);

    changeLogOldestOffsets =
        getChangeLogOldestOffsetsForPartition(taskModel.getChangelogPartition(), changeLogMetadata);
    LOG.info("Assigning oldest change log offsets for taskName {} : {}", taskModel.getTaskName(),
        changeLogOldestOffsets);
  }
Get the oldest offset for each changelog SSP based on the stream's metadata (obtained from streamMetadataCache).
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java