private static Set<SystemStreamPartition> getInputStreamPartitions(Config config,
      StreamMetadataCache streamMetadataCache) {
    TaskConfig taskConfig = new TaskConfig(config);
    // Get the set of partitions for each SystemStream from the stream metadata
    Map<SystemStream, SystemStreamMetadata> allMetadata = JavaConverters.mapAsJavaMapConverter(
        streamMetadataCache.getStreamMetadata(
            JavaConverters.asScalaSetConverter(taskConfig.getInputStreams()).asScala().toSet(), true)).asJava();
    Set<SystemStreamPartition> inputStreamPartitions = new HashSet<>();
    allMetadata.forEach((systemStream, systemStreamMetadata) -> systemStreamMetadata.getSystemStreamPartitionMetadata()
        .keySet()
        .forEach(partition -> inputStreamPartitions.add(new SystemStreamPartition(systemStream, partition))));
    return inputStreamPartitions;
  }
Computes the input system stream partitions of a samza job using the provided {@param config}
and {@param streamMetadataCache}.
@param config the configuration of the job.
@param streamMetadataCache to query the partition metadata of the input streams.
@return the input {@see SystemStreamPartition} of the samza job.
samza-core/src/main/java/org/apache/samza/coordinator/JobModelCalculator.java