@Test
  public void testJobModelExpiredContainerShutdownTimeout() throws InterruptedException {
    JobCoordinator mockJobCoordinator = mock(JobCoordinator.class);
    // use this to store the exception passed to afterFailure for the processor lifecycle listener
    AtomicReference<Throwable> afterFailureException = new AtomicReference<>(null);
    TestableStreamProcessor processor = new TestableStreamProcessor(
        // set a small shutdown timeout so it triggers faster
        new MapConfig(ImmutableMap.of(TaskConfig.TASK_SHUTDOWN_MS, "1")),
        new HashMap<>(),
        mock(StreamTaskFactory.class),
        new ProcessorLifecycleListener() {
          @Override
          public void beforeStart() { }

          @Override
          public void afterStart() { }

          @Override
          public void afterFailure(Throwable t) {
            afterFailureException.set(t);
          }

          @Override
          public void afterStop() { }
        },
        mockJobCoordinator,
        null,
        // take an extra second to shut down so that task shutdown timeout gets reached
        Duration.of(1, ChronoUnit.SECONDS));

    Thread jcThread = new Thread(() -> {
      // gets processor into rebalance mode so onNewJobModel creates a new container
      processor.jobCoordinatorListener.onJobModelExpired();
      processor.jobCoordinatorListener.onNewJobModel("1", getMockJobModel());
      try {
        // wait for the run loop to be ready before triggering rebalance
        processor.runLoopStartForMain.await();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      processor.jobCoordinatorListener.onJobModelExpired();
    });
    doAnswer(invocation -> {
      jcThread.start();
      return null;
    }).when(mockJobCoordinator).start();

    // ensure that the coordinator stop occurred before checking the exception being thrown
    CountDownLatch coordinatorStop = new CountDownLatch(1);
    doAnswer(invocation -> {
      processor.jobCoordinatorListener.onCoordinatorStop();
      coordinatorStop.countDown();
      return null;
    }).when(mockJobCoordinator).stop();

    processor.start();

    // make sure the job model expired callback completed
    assertTrue("Job coordinator stop not called", coordinatorStop.await(10, TimeUnit.SECONDS));
    assertNotNull(afterFailureException.get());
    assertTrue(afterFailureException.get() instanceof TimeoutException);
  }
Given that the job model expires, but the container takes too long to stop, a TimeoutException should be propagated
to the processor lifecycle listener.
samza-core/src/test/java/org/apache/samza/processor/TestStreamProcessor.java