@Test
  public void testStreamProcessorWithStreamProcessorListenerFactory() {
    AtomicReference<MockStreamProcessorLifecycleListener> mockListener = new AtomicReference<>();
    StreamProcessor streamProcessor =
        new StreamProcessor("TestProcessorId", mock(Config.class), new HashMap<>(), mock(TaskFactory.class),
            Optional.empty(), Optional.empty(), Optional.empty(), sp ->
            mockListener.updateAndGet(old -> new MockStreamProcessorLifecycleListener(sp)),
            mock(JobCoordinator.class), Mockito.mock(MetadataStore.class));
    assertEquals(streamProcessor, mockListener.get().processor);
  }
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