StreamSpec getStreamSpec() {
    StreamSpec spec = (partitions == PARTITIONS_UNKNOWN) ?
        streamSpec : streamSpec.copyWithPartitionCount(partitions);

    // For intermediate stream that physical name is the same as id,
    // meaning the physical name is auto-generated, and not overrided
    // by user or batch processing.
    if (isIntermediate && spec.getId().equals(spec.getPhysicalName())) {
      // Append unique id to the batch intermediate streams
      // Note this will only happen for batch processing
      String physicalName = StreamManager.createUniqueNameForBatch(spec.getPhysicalName(), config);
      if (!physicalName.equals(spec.getPhysicalName())) {
        spec = spec.copyWithPhysicalName(physicalName);
      }
    }
    return spec;
  }
A StreamEdge connects the source {@link JobNode}s to the target {@link JobNode}s with a stream.
If it's a sink StreamEdge, the target JobNode is empty.
If it's a source StreamEdge, the source JobNode is empty.
samza-core/src/main/java/org/apache/samza/execution/StreamEdge.java