private Map<SystemStream, String> getChangeLogOldestOffsetsForPartition(Partition partition,
      Map<SystemStream, SystemStreamMetadata> inputStreamMetadata) {

    Map<SystemStream, String> retVal = new HashMap<>();

    // NOTE: do not use Collectors.Map because of https://bugs.openjdk.java.net/browse/JDK-8148463
    inputStreamMetadata.entrySet()
        .stream()
        .filter(x -> x.getValue().getSystemStreamPartitionMetadata().get(partition) != null)
        .forEach(e -> retVal.put(e.getKey(),
            e.getValue().getSystemStreamPartitionMetadata().get(partition).getOldestOffset()));

    return retVal;
  }
Builds a map from SystemStreamPartition to oldest offset for changelogs.
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java