@Test
  public void testFetchExistingStreamPartitions() {
    ExecutionPlanner planner = new ExecutionPlanner(config, streamManager);
    StreamApplicationDescriptorImpl graphSpec = createStreamGraphWithStreamStreamJoin();
    JobGraph jobGraph = planner.createJobGraph(graphSpec);

    planner.setInputAndOutputStreamPartitionCount(jobGraph);
    assertTrue(jobGraph.getOrCreateStreamEdge(input1Spec).getPartitionCount() == 64);
    assertTrue(jobGraph.getOrCreateStreamEdge(input2Spec).getPartitionCount() == 16);
    assertTrue(jobGraph.getOrCreateStreamEdge(input3Spec).getPartitionCount() == 32);
    assertTrue(jobGraph.getOrCreateStreamEdge(output1Spec).getPartitionCount() == 8);
    assertTrue(jobGraph.getOrCreateStreamEdge(output2Spec).getPartitionCount() == 16);

    jobGraph.getIntermediateStreamEdges().forEach(edge -> {
      assertTrue(edge.getPartitionCount() == -1);
    });
  }
A special example of stream-table join where a stream is joined with a table, and the result is
sent to the same table. This example is necessary to ensure {@link ExecutionPlanner} does not
get stuck traversing the virtual cycle between stream-table-join and send-to-table operator specs
indefinitely.

The reason such virtual cycle is present is to support computing partitions of intermediate
streams participating in stream-table joins. Please, refer to SAMZA SEP-16 for more details.
samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java