@Override
      public void onJobModelExpired() {
        synchronized (lock) {
          if (state == State.STARTED || state == State.RUNNING) {
            state = State.IN_REBALANCE;
            LOGGER.info("Job model expired. Shutting down the container: {} of stream processor: {}.", container, processorId);
            boolean hasContainerShutdown = stopSamzaContainer();
            if (!hasContainerShutdown) {
              LOGGER.warn("Container: {} shutdown was unsuccessful. Stopping the stream processor: {}.", container, processorId);
              state = State.STOPPING;
              jobCoordinator.stop();
            } else {
              LOGGER.info("Container: {} shutdown completed for stream processor: {}.", container, processorId);
            }
          } else if (state == State.IN_REBALANCE) {
            if (container != null) {
              boolean hasContainerShutdown = interruptContainerAndShutdownExecutorService();
              if (!hasContainerShutdown) {
                LOGGER.warn("Job model expire unsuccessful. Failed to interrupt container: {} safely. "
                    + "Stopping the stream processor: {}", container, processorId);
                state = State.STOPPING;
                jobCoordinator.stop();
              } else {
                containerExecutorService = createExecutorService();
              }
            } else {
              LOGGER.info("Ignoring Job model expired since a rebalance is already in progress");
            }
          } else {
            LOGGER.info("Ignoring onJobModelExpired invocation since the current state is {} and not in {}.", state,
                ImmutableList.of(State.RUNNING, State.STARTED, State.IN_REBALANCE));
          }
        }
      }
Stops the {@link SamzaContainer}.
@return true if {@link SamzaContainer} had shutdown within task.shutdown.ms. false otherwise.
samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java