@Test
  public void testContainerFailureCorrectlyStopsProcessor() throws InterruptedException {
    JobCoordinator mockJobCoordinator = Mockito.mock(JobCoordinator.class);
    Throwable expectedThrowable =  new SamzaException("Failure in Container!");
    AtomicReference<Throwable> actualThrowable = new AtomicReference<>();
    final CountDownLatch runLoopStartedLatch = new CountDownLatch(1);
    RunLoop failingRunLoop = mock(RunLoop.class);
    doAnswer(invocation -> {
      try {
        runLoopStartedLatch.countDown();
        throw expectedThrowable;
      } catch (InterruptedException ie) {
        ie.printStackTrace();
      }
      return null;
    }).when(failingRunLoop).run();

    SamzaContainer mockContainer = StreamProcessorTestUtils.getDummyContainer(failingRunLoop, mock(StreamTask.class));
    final CountDownLatch processorListenerFailed = new CountDownLatch(1);

    TestableStreamProcessor processor = new TestableStreamProcessor(
        new MapConfig(),
        new HashMap<>(),
        mock(StreamTaskFactory.class),
        new ProcessorLifecycleListener() {
          @Override
          public void beforeStart() {
            processorListenerState.put(ListenerCallback.BEFORE_START, true);
          }

          @Override
          public void afterStart() {
            processorListenerState.put(ListenerCallback.AFTER_START, true);
          }

          @Override
          public void afterStop() {
            processorListenerState.put(ListenerCallback.AFTER_STOP, true);
          }

          @Override
          public void afterFailure(Throwable t) {
            processorListenerState.put(ListenerCallback.AFTER_FAILURE, true);
            actualThrowable.getAndSet(t);
            processorListenerFailed.countDown();
          }
        },
        mockJobCoordinator,
        mockContainer);

    final CountDownLatch coordinatorStop = new CountDownLatch(1);
    doAnswer(invocation -> {
      coordinatorStop.countDown();
      return null;
    }).when(mockJobCoordinator).stop();

    doAnswer(invocation -> {
      new Thread(() -> {
        try {
          processor.jobCoordinatorListener.onJobModelExpired();
          processor.jobCoordinatorListener.onNewJobModel("1", getMockJobModel());
          coordinatorStop.await();
          processor.jobCoordinatorListener.onCoordinatorStop();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }).start();
      return null;
    }).when(mockJobCoordinator).start();

    processor.start();

    // This block is required for the mockRunloop is actually started.
    // Otherwise, processor.stop gets triggered before mockRunloop begins to block
    runLoopStartedLatch.await();
    assertTrue(
        "Container failed and processor listener failed was not invoked within timeout!",
        processorListenerFailed.await(30, TimeUnit.SECONDS));
    assertEquals(expectedThrowable, actualThrowable.get());

    assertTrue(processorListenerState.get(ListenerCallback.BEFORE_START));
    assertTrue(processorListenerState.get(ListenerCallback.AFTER_START));
    Assert.assertFalse(processorListenerState.get(ListenerCallback.AFTER_STOP));
    assertTrue(processorListenerState.get(ListenerCallback.AFTER_FAILURE));
  }
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