@Before
  public void setup() {
    Map<String, String> configMap = new HashMap<>();
    configMap.put(JobConfig.JOB_NAME, "test-app");
    configMap.put(JobConfig.JOB_DEFAULT_SYSTEM, DEFAULT_SYSTEM);
    StreamTestUtils.addStreamConfigs(configMap, "input1", "system1", "input1");
    StreamTestUtils.addStreamConfigs(configMap, "input2", "system2", "input2");
    StreamTestUtils.addStreamConfigs(configMap, "input3", "system2", "input3");
    StreamTestUtils.addStreamConfigs(configMap, "input4", "system1", "input4");
    StreamTestUtils.addStreamConfigs(configMap, "output1", "system1", "output1");
    StreamTestUtils.addStreamConfigs(configMap, "output2", "system2", "output2");
    config = new MapConfig(configMap);

    input1Spec = new StreamSpec("input1", "input1", "system1");
    input2Spec = new StreamSpec("input2", "input2", "system2");
    input3Spec = new StreamSpec("input3", "input3", "system2");

    output1Spec = new StreamSpec("output1", "output1", "system1");
    output2Spec = new StreamSpec("output2", "output2", "system2");

    KVSerde<Object, Object> kvSerde = new KVSerde<>(new NoOpSerde(), new NoOpSerde());
    String mockSystemFactoryClass = "factory.class.name";
    system1Descriptor = new GenericSystemDescriptor("system1", mockSystemFactoryClass);
    system2Descriptor = new GenericSystemDescriptor("system2", mockSystemFactoryClass);
    input1Descriptor = system1Descriptor.getInputDescriptor("input1", kvSerde);
    input2Descriptor = system2Descriptor.getInputDescriptor("input2", kvSerde);
    input3Descriptor = system2Descriptor.getInputDescriptor("input3", kvSerde);
    input4Descriptor = system1Descriptor.getInputDescriptor("input4", kvSerde);
    output1Descriptor = system1Descriptor.getOutputDescriptor("output1", kvSerde);
    output2Descriptor = system2Descriptor.getOutputDescriptor("output2", kvSerde);

    // clean and set up sets and maps of descriptors
    systemDescriptors.clear();
    inputDescriptors.clear();
    outputDescriptors.clear();
    tableDescriptors.clear();
    systemDescriptors.add(system1Descriptor);
    systemDescriptors.add(system2Descriptor);
    inputDescriptors.put(input1Descriptor.getStreamId(), input1Descriptor);
    inputDescriptors.put(input2Descriptor.getStreamId(), input2Descriptor);
    inputDescriptors.put(input3Descriptor.getStreamId(), input3Descriptor);
    inputDescriptors.put(input4Descriptor.getStreamId(), input4Descriptor);
    outputDescriptors.put(output1Descriptor.getStreamId(), output1Descriptor);
    outputDescriptors.put(output2Descriptor.getStreamId(), output2Descriptor);


    // set up external partition count
    Map<String, Integer> system1Map = new HashMap<>();
    system1Map.put("input1", 64);
    system1Map.put("output1", 8);
    system1Map.put("input4", IntermediateStreamManager.MAX_INFERRED_PARTITIONS * 2);
    Map<String, Integer> system2Map = new HashMap<>();
    system2Map.put("input2", 16);
    system2Map.put("input3", 32);
    system2Map.put("output2", 16);

    SystemAdmin systemAdmin1 = createSystemAdmin(system1Map);
    SystemAdmin systemAdmin2 = createSystemAdmin(system2Map);
    systemAdmins = mock(SystemAdmins.class);
    when(systemAdmins.getSystemAdmin("system1")).thenReturn(systemAdmin1);
    when(systemAdmins.getSystemAdmin("system2")).thenReturn(systemAdmin2);
    streamManager = new StreamManager(systemAdmins);
  }
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