@Test
  public void testStopShouldBeIdempotent() {
    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 = PowerMockito.spy(new StreamProcessor("TestProcessorId", config, new HashMap<>(),
        null, Optional.empty(), Optional.empty(), Optional.empty(), sp -> lifecycleListener, mockJobCoordinator,
        Mockito.mock(MetadataStore.class)));

    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.state = State.RUNNING;

    streamProcessor.stop();

    assertEquals(State.STOPPING, streamProcessor.state);
  }
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