@Test
  public void testTriggerIntervalWhenWindowMsIsConfigured() {
    Map<String, String> map = new HashMap<>(config);
    map.put(TaskConfig.WINDOW_MS, "2000");
    map.put(JobConfig.JOB_INTERMEDIATE_STREAM_PARTITIONS, String.valueOf(DEFAULT_PARTITIONS));
    Config cfg = new MapConfig(map);

    ExecutionPlanner planner = new ExecutionPlanner(cfg, streamManager);
    StreamApplicationDescriptorImpl graphSpec = createSimpleGraph();
    ExecutionPlan plan = planner.plan(graphSpec);
    List<JobConfig> jobConfigs = plan.getJobConfigs();
    assertEquals(1, jobConfigs.size());
    assertEquals("2000", jobConfigs.get(0).get(TaskConfig.WINDOW_MS));
  }
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