@Test
  public void testStreamCreation()
      throws Exception {
    remotePlanner = createRemoteJobPlanner(mock(StreamApplicationDescriptorImpl.class));
    StreamManager streamManager = mock(StreamManager.class);
    doReturn(streamManager).when(remotePlanner).buildAndStartStreamManager(any(Config.class));

    ExecutionPlan plan = mock(ExecutionPlan.class);
    when(plan.getIntermediateStreams()).thenReturn(
        Collections.singletonList(new StreamSpec("test-stream", "test-stream", "test-system")));
    when(plan.getPlanAsJson()).thenReturn("");
    when(plan.getJobConfigs()).thenReturn(Collections.singletonList(mock(JobConfig.class)));
    ApplicationConfig mockAppConfig = mock(ApplicationConfig.class);
    when(mockAppConfig.getAppMode()).thenReturn(ApplicationConfig.ApplicationMode.STREAM);
    when(plan.getApplicationConfig()).thenReturn(mockAppConfig);
    doReturn(plan).when(remotePlanner).getExecutionPlan(any());

    remotePlanner.prepareJobs();

    verify(streamManager, times(0)).clearStreamsFromPreviousRun(any());
    ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
    verify(streamManager).createStreams(captor.capture());
    List<StreamSpec> streamSpecs = captor.getValue();
    assertEquals(streamSpecs.size(), 1);
    assertEquals(streamSpecs.get(0).getId(), "test-stream");
    verify(streamManager).stop();
  }
Unit tests for {@link RemoteJobPlanner}

TODO: consolidate this with unit tests for ExecutionPlanner after SAMZA-1811
samza-core/src/test/java/org/apache/samza/execution/TestRemoteJobPlanner.java