private SystemStreamMetadata constructSystemStreamMetadata(
      String streamName,
      Map<SystemStreamPartition, List<IncomingMessageEnvelope>> sspToMessagesForSystem) {

    Map<Partition, SystemStreamMetadata.SystemStreamPartitionMetadata> partitionMetadata =
        sspToMessagesForSystem
            .entrySet()
            .stream()
            .collect(Collectors.toMap(entry -> entry.getKey().getPartition(), entry -> {
              List<IncomingMessageEnvelope> messages = entry.getValue();
              Integer oldestOffset;
              Integer newestOffset;
              int upcomingOffset;

              if (messages.isEmpty()) {
                oldestOffset = null;
                newestOffset = null;
                upcomingOffset = 0;
              } else if (messages.get(messages.size() - 1).isEndOfStream()) {
                if (messages.size() > 1) {
                  // don't count end of stream in offset indices
                  oldestOffset = 0;
                  newestOffset = messages.size() - 2;
                  upcomingOffset = messages.size() - 1;
                } else {
                  // end of stream is the only message, treat the same as empty
                  oldestOffset = null;
                  newestOffset = null;
                  upcomingOffset = 0;
                }
              } else {
                // offsets correspond strictly to numeric indices
                oldestOffset = 0;
                newestOffset = messages.size() - 1;
                upcomingOffset = messages.size();
              }

              return new SystemStreamMetadata.SystemStreamPartitionMetadata(
                  oldestOffset == null ? null : oldestOffset.toString(),
                  newestOffset == null ? null : newestOffset.toString(),
                  Integer.toString(upcomingOffset));
            }));

    return new SystemStreamMetadata(streamName, partitionMetadata);
  }
Fetch partition count for the given input stream stream.

@param systemStream input system stream

@return the partition count if available or {@link InMemoryManager#DEFAULT_PARTITION_COUNT}
samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java