private void runJobModelExpireDuringRebalance(StreamProcessor streamProcessor, ExecutorService executorService,
      boolean failContainerInterrupt) {
    SamzaContainer mockSamzaContainer = Mockito.mock(SamzaContainer.class);
    CountDownLatch shutdownLatch = new CountDownLatch(1);

    /*
     * When there is an initialized container that hasn't started and the stream processor is still in re-balance phase,
     * subsequent job model expire request should attempt to interrupt the existing container to safely shut it down
     * before proceeding to join the barrier. As part of safe shutdown sequence,  we want to ensure shutdownNow is invoked
     * on the existing executorService to signal interrupt and make sure new executor service is created.
     */

    Mockito.when(executorService.shutdownNow()).thenAnswer(ctx -> {
      if (!failContainerInterrupt) {
        shutdownLatch.countDown();
      }
      return null;
    });
    Mockito.when(executorService.isShutdown()).thenReturn(true);

    streamProcessor.state = State.IN_REBALANCE;
    streamProcessor.container = mockSamzaContainer;
    streamProcessor.containerExecutorService = executorService;
    streamProcessor.containerShutdownLatch = shutdownLatch;

    streamProcessor.jobCoordinatorListener.onJobModelExpired();
  }
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