@Test
  public void testStopByProcessor() throws InterruptedException {
    JobCoordinator mockJobCoordinator = mock(JobCoordinator.class);

    final CountDownLatch processorListenerStop = new CountDownLatch(1);
    final CountDownLatch processorListenerStart = new CountDownLatch(1);

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

          @Override
          public void afterFailure(Throwable t) {
            processorListenerState.put(ListenerCallback.AFTER_FAILURE, true);
          }

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

          @Override
          public void beforeStart() {
            processorListenerState.put(ListenerCallback.BEFORE_START, true);
          }
        },
        mockJobCoordinator,
        null);

    final CountDownLatch coordinatorStop = new CountDownLatch(1);
    final Thread jcThread = new Thread(() -> {
      try {
        processor.jobCoordinatorListener.onJobModelExpired();
        processor.jobCoordinatorListener.onNewJobModel("1", getMockJobModel());
        coordinatorStop.await();
        processor.jobCoordinatorListener.onCoordinatorStop();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    });

    doAnswer(invocation -> {
      coordinatorStop.countDown();
      return null;
    }).when(mockJobCoordinator).stop();

    doAnswer(invocation -> {
      jcThread.start();
      return null;
    }).when(mockJobCoordinator).start();

    processor.start();
    processorListenerStart.await(10, TimeUnit.SECONDS);

    assertEquals(SamzaContainerStatus.STARTED, processor.getContainerStatus());

    // This block is required for the mockRunloop is actually start.
    // Otherwise, processor.stop gets triggered before mockRunloop begins to block
    processor.runLoopStartForMain.await();

    processor.stop();

    processorListenerStop.await();

    // Assertions on which callbacks are expected to be invoked
    assertTrue(processorListenerState.get(ListenerCallback.BEFORE_START));
    assertTrue(processorListenerState.get(ListenerCallback.AFTER_START));
    assertTrue(processorListenerState.get(ListenerCallback.AFTER_STOP));
    Assert.assertFalse(processorListenerState.get(ListenerCallback.AFTER_FAILURE));
  }
Tests stop() method when Container AND JobCoordinator are running
samza-core/src/test/java/org/apache/samza/processor/TestStreamProcessor.java