@Test
  public void testStreamCreationWithCoordination()
      throws Exception {
    StreamApplicationDescriptorImpl appDesc = mock(StreamApplicationDescriptorImpl.class);
    doReturn(mock(Config.class)).when(appDesc).getConfig();
    localPlanner = createLocalJobPlanner(appDesc);
    StreamManager streamManager = mock(StreamManager.class);
    doReturn(streamManager).when(localPlanner).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)));
    doReturn(plan).when(localPlanner).getExecutionPlan(any());

    CoordinationUtils coordinationUtils = mock(CoordinationUtils.class);
    CoordinationUtilsFactory coordinationUtilsFactory = mock(CoordinationUtilsFactory.class);
    JobCoordinatorConfig mockJcConfig = mock(JobCoordinatorConfig.class);
    when(mockJcConfig.getCoordinationUtilsFactory()).thenReturn(coordinationUtilsFactory);
    PowerMockito.whenNew(JobCoordinatorConfig.class).withAnyArguments().thenReturn(mockJcConfig);

    DistributedLock lock = mock(DistributedLock.class);
    when(lock.lock(anyObject())).thenReturn(true);
    when(coordinationUtils.getLock(anyString())).thenReturn(lock);
    when(coordinationUtilsFactory.getCoordinationUtils(anyString(), anyString(), anyObject()))
        .thenReturn(coordinationUtils);

    localPlanner.prepareJobs();

    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 LocalJobPlanner}

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