@VisibleForTesting
  Map<SystemStreamPartition, String> getOldestOffsets() {
    Map<SystemStreamPartition, String> oldestOffsets = new HashMap<>();

    // Step 1
    Map<SystemStream, List<SystemStreamPartition>> systemStreamToSsp = this.sspToStores.keySet().stream()
        .collect(Collectors.groupingBy(SystemStreamPartition::getSystemStream));

    // Step 2
    Map<SystemStream, SystemStreamMetadata> metadata = JavaConverters.mapAsJavaMapConverter(
        this.streamMetadataCache.getStreamMetadata(
            JavaConverters.asScalaSetConverter(systemStreamToSsp.keySet()).asScala().toSet(), false)).asJava();

    // Step 3
    metadata.forEach((systemStream, systemStreamMetadata) -> {

      // get the partition metadata for each system stream
      Map<Partition, SystemStreamMetadata.SystemStreamPartitionMetadata> partitionMetadata =
          systemStreamMetadata.getSystemStreamPartitionMetadata();

      // For SSPs belonging to the system stream, use the partition metadata to get the oldest offset
      // if partitionMetadata was not obtained for any SSP, populate oldest-offset as null
      // Because of https://bugs.openjdk.java.net/browse/JDK-8148463 using lambda will NPE when getOldestOffset() is null
      for (SystemStreamPartition ssp : systemStreamToSsp.get(systemStream)) {
        oldestOffsets.put(ssp, partitionMetadata.get(ssp.getPartition()).getOldestOffset());
      }
    });

    return oldestOffsets;
  }
Gets the oldest offset for the {@link SystemStreamPartition}s associated with all the store side inputs.
  1. Groups the list of the SSPs based on system stream
  2. Fetches the {@link SystemStreamMetadata} from {@link StreamMetadataCache}
  3. Fetches the partition metadata for each system stream and fetch the corresponding partition metadata
     and populates the oldest offset for SSPs belonging to the system stream.

@return a {@link Map} of {@link SystemStreamPartition} to their oldest offset. If partitionMetadata could not be
obtained for any {@link SystemStreamPartition} the offset for it is populated as null.
samza-core/src/main/java/org/apache/samza/storage/TaskSideInputHandler.java