@Test
  public void testCoordinatorStopShouldStopTheStreamProcessor() {
    JobCoordinator mockJobCoordinator = Mockito.mock(JobCoordinator.class);
    ProcessorLifecycleListener lifecycleListener = Mockito.mock(ProcessorLifecycleListener.class);
    MapConfig config = new MapConfig(ImmutableMap.of("task.shutdown.ms", "0"));
    StreamProcessor streamProcessor = new StreamProcessor("TestProcessorId", config, new HashMap<>(), null,
        Optional.empty(), Optional.empty(), Optional.empty(), sp -> lifecycleListener, mockJobCoordinator, Mockito.mock(MetadataStore.class));

    streamProcessor.state = State.RUNNING;
    streamProcessor.jobCoordinatorListener.onCoordinatorStop();

    assertEquals(State.STOPPED, streamProcessor.state);
    Mockito.verify(lifecycleListener).afterStop();
  }
When there's initialized SamzaContainer in StreamProcessor and the container shutdown
fails in onJobModelExpired. onJobModelExpired should move StreamProcessor to STOPPING
state and should shutdown JobCoordinator.
samza-core/src/test/java/org/apache/samza/processor/TestStreamProcessor.java