private Map<SystemStreamPartition, SystemStreamMetadata.SystemStreamPartitionMetadata> getInitialSideInputSSPMetadata() {
    Map<SystemStreamPartition, SystemStreamMetadata.SystemStreamPartitionMetadata> initialSideInputSSPMetadata = new HashMap<>();
    for (SystemStreamPartition ssp : this.sspToStores.keySet()) {
      boolean partitionsMetadataOnly = false;
      SystemStreamMetadata systemStreamMetadata = this.streamMetadataCache.getSystemStreamMetadata(ssp.getSystemStream(), partitionsMetadataOnly);
      if (systemStreamMetadata != null) {
        SystemStreamMetadata.SystemStreamPartitionMetadata sspMetadata =
            systemStreamMetadata.getSystemStreamPartitionMetadata().get(ssp.getPartition());
        initialSideInputSSPMetadata.put(ssp, sspMetadata);
      }
    }
    return initialSideInputSSPMetadata;
  }
Retrieves the newest offset for each SSP

@return a map of SSP to metadata
samza-core/src/main/java/org/apache/samza/storage/TaskSideInputHandler.java