private static List<StreamSet> groupJoinedStreams(JobGraph jobGraph) {
    // Group input operator specs (input/intermediate streams) by the joins they participate in.
    Multimap<OperatorSpec, InputOperatorSpec> joinOpSpecToInputOpSpecs =
        OperatorSpecGraphAnalyzer.getJoinToInputOperatorSpecs(
            jobGraph.getApplicationDescriptorImpl().getInputOperators().values());

    Map<String, TableDescriptor> tableDescriptors = jobGraph.getTables().stream()
        .collect(Collectors.toMap(TableDescriptor::getTableId, Function.identity()));

    // Convert every group of input operator specs into a group of corresponding stream edges.
    List<StreamSet> streamSets = new ArrayList<>();
    for (OperatorSpec joinOpSpec : joinOpSpecToInputOpSpecs.keySet()) {
      Collection<InputOperatorSpec> joinedInputOpSpecs = joinOpSpecToInputOpSpecs.get(joinOpSpec);
      StreamSet streamSet = getStreamSet(joinOpSpec.getOpId(), joinedInputOpSpecs, jobGraph);

      // If current join is a stream-table join, add the stream edges corresponding to side-input
      // streams associated with the joined table (if any).
      if (joinOpSpec instanceof StreamTableJoinOperatorSpec) {
        StreamTableJoinOperatorSpec streamTableJoinOperatorSpec = (StreamTableJoinOperatorSpec) joinOpSpec;
        TableDescriptor tableDescriptor = tableDescriptors.get(streamTableJoinOperatorSpec.getTableId());
        if (tableDescriptor instanceof LocalTableDescriptor) {
          LocalTableDescriptor localTableDescriptor = (LocalTableDescriptor) tableDescriptor;
          Collection<String> sideInputs = ListUtils.emptyIfNull(localTableDescriptor.getSideInputs());
          Iterable<StreamEdge> sideInputStreams = sideInputs.stream().map(jobGraph::getStreamEdge)::iterator;
          Iterable<StreamEdge> streams = streamSet.getStreamEdges();
          streamSet = new StreamSet(streamSet.getSetId(), Iterables.concat(streams, sideInputStreams));
        }
      }

      streamSets.add(streamSet);
    }

    return Collections.unmodifiableList(streamSets);
  }
Groups streams participating in joins together.
samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java