private Set<SystemStreamPartition> getInputStreamPartitions() {
    TaskConfig taskConfig = new TaskConfig(config);
    scala.collection.immutable.Set<SystemStream> inputSystemStreams =
        JavaConverters.asScalaSetConverter(taskConfig.getInputStreams()).asScala().toSet();

    // Get the set of partitions for each SystemStream from the stream metadata
    Set<SystemStreamPartition>
        sspSet = JavaConverters.mapAsJavaMapConverter(streamMetadataCache.getStreamMetadata(inputSystemStreams, true)).asJava()
        .entrySet()
        .stream()
        .flatMap(this::mapSSMToSSP)
        .collect(Collectors.toSet());

    return sspSet;
  }
For each input stream specified in config, exactly determine its
partitions, returning a set of SystemStreamPartitions containing them all.
samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java