@Test
  public void testOnJobModelExpiredShouldMakeCorrectStateTransitions() {
    JobCoordinator mockJobCoordinator = Mockito.mock(JobCoordinator.class);
    ProcessorLifecycleListener lifecycleListener = Mockito.mock(ProcessorLifecycleListener.class);
    SamzaContainer mockSamzaContainer = Mockito.mock(SamzaContainer.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));

    /**
     * Without a SamzaContainer running in StreamProcessor and current StreamProcessor state is STARTED,
     * onJobModelExpired should move the state to IN_REBALANCE.
     */

    streamProcessor.start();

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

    streamProcessor.jobCoordinatorListener.onJobModelExpired();

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

    /**
     * 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.
     */
    Mockito.doNothing().when(mockJobCoordinator).start();
    Mockito.doNothing().when(mockJobCoordinator).stop();
    Mockito.doNothing().when(mockSamzaContainer).shutdown();
    Mockito.when(mockSamzaContainer.hasStopped()).thenReturn(false);
    Mockito.when(mockSamzaContainer.getStatus())
            .thenReturn(SamzaContainerStatus.STARTED)
            .thenReturn(SamzaContainerStatus.STOPPED);
    streamProcessor.container = mockSamzaContainer;
    streamProcessor.state = State.STARTED;

    streamProcessor.jobCoordinatorListener.onJobModelExpired();

    assertEquals(State.STOPPING, streamProcessor.getState());
    Mockito.verify(mockSamzaContainer, Mockito.times(1)).shutdown();
    Mockito.verify(mockJobCoordinator, Mockito.times(1)).stop();
  }
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