@VisibleForTesting
  static Map<String, SystemStreamMetadata> assembleMetadata(Map<SystemStreamPartition, String> oldestOffsets,
      Map<SystemStreamPartition, String> newestOffsets, Map<SystemStreamPartition, String> upcomingOffsets) {
    HashSet<SystemStreamPartition> allSSPs = new HashSet<>();
    allSSPs.addAll(oldestOffsets.keySet());
    allSSPs.addAll(newestOffsets.keySet());
    allSSPs.addAll(upcomingOffsets.keySet());

    Map<String, SystemStreamMetadata> assembledMetadata = allSSPs.stream()
        .collect(Collectors.groupingBy(SystemStreamPartition::getStream))
        .entrySet()
        .stream()
        .collect(Collectors.toMap(Map.Entry::getKey, entry -> {
          Map<Partition, SystemStreamMetadata.SystemStreamPartitionMetadata> partitionMetadata =
              entry.getValue()
                  .stream()
                  .collect(Collectors.toMap(SystemStreamPartition::getPartition, ssp ->
                      new SystemStreamMetadata.SystemStreamPartitionMetadata(
                          oldestOffsets.getOrDefault(ssp, null),
                          newestOffsets.getOrDefault(ssp, null),
                          upcomingOffsets.get(ssp))));
          return new SystemStreamMetadata(entry.getKey(), partitionMetadata);
        }));

    return assembledMetadata;
  }
A helper method that takes oldest, newest, and upcoming offsets for each
system stream partition, and creates a single map from stream name to
SystemStreamMetadata.

@param newestOffsets map of SSP to newest offset
@param oldestOffsets map of SSP to oldest offset
@param upcomingOffsets map of SSP to upcoming offset
@return a {@link Map} from {@code system} to {@link SystemStreamMetadata}
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java