@Test
  public void testMaxPartitionLimitIsIgnoredForStreamingMode() {
    int expectedPartitionSize = IntermediateStreamManager.MAX_INFERRED_PARTITIONS * 2;

    ExecutionPlanner planner = new ExecutionPlanner(config, streamManager);
    StreamApplicationDescriptorImpl graphSpec = new StreamApplicationDescriptorImpl(appDesc -> {
      MessageStream<KV<Object, Object>> input1 = appDesc.getInputStream(input4Descriptor);
      OutputStream<KV<Object, Object>> output1 = appDesc.getOutputStream(output1Descriptor);
      input1.partitionBy(m -> m.key, m -> m.value, mock(KVSerde.class), "p1").map(kv -> kv).sendTo(output1);
    }, config);

    JobGraph jobGraph = (JobGraph) planner.plan(graphSpec);

    // Partitions should be the same as input1
    jobGraph.getIntermediateStreams().forEach(edge -> {
      assertEquals(expectedPartitionSize, edge.getPartitionCount()); // max of input1 and output1
    });
  }
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