@Test
  public void testStartOperationShouldBeIdempotent() {
    JobCoordinator mockJobCoordinator = Mockito.mock(JobCoordinator.class);
    Mockito.doNothing().when(mockJobCoordinator).start();
    ProcessorLifecycleListener lifecycleListener = Mockito.mock(ProcessorLifecycleListener.class);
    StreamProcessor streamProcessor = Mockito.spy(new StreamProcessor("TestProcessorId", new MapConfig(),
        new HashMap<>(), null, Optional.empty(), Optional.empty(), Optional.empty(), sp -> lifecycleListener,
        mockJobCoordinator, Mockito.mock(MetadataStore.class)));
    assertEquals(State.NEW, streamProcessor.getState());
    streamProcessor.start();

    assertEquals(State.STARTED, streamProcessor.getState());

    streamProcessor.start();

    assertEquals(State.STARTED, streamProcessor.getState());

    Mockito.verify(mockJobCoordinator, Mockito.times(1)).start();
  }
Tests that a failure in container correctly stops a running JobCoordinator and propagates the exception
through the StreamProcessor

Assertions:
- JobCoordinator has been stopped from the JobCoordinatorListener callback
- ProcessorLifecycleListener#afterStop(Throwable) has been invoked w/ non-null Throwable
samza-core/src/test/java/org/apache/samza/processor/TestStreamProcessor.java