@Test
  public void testOnNewJobModelShouldResultInValidStateTransitions() throws Exception {
    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 TestableStreamProcessor(config, new HashMap<>(), null,
        lifecycleListener, mockJobCoordinator, mockSamzaContainer);

    streamProcessor.state = State.IN_REBALANCE;
    Mockito.doNothing().when(mockSamzaContainer).run();

    streamProcessor.jobCoordinatorListener.onNewJobModel("TestProcessorId", new JobModel(new MapConfig(), new HashMap<>()));

    Mockito.verify(mockSamzaContainer, Mockito.times(1)).setContainerListener(any());
    Mockito.verify(mockSamzaContainer, Mockito.atMost(1)).run();
  }
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