@Override
      public void onCoordinatorStop() {
        synchronized (lock) {
          LOGGER.info("Shutting down the executor service of the stream processor: {}.", processorId);
          boolean hasContainerShutdown = stopSamzaContainer();

          // we only want to interrupt when container shutdown times out.
          if (!hasContainerShutdown) {
            containerExecutorService.shutdownNow();
          }
          state = State.STOPPED;
        }
        if (containerException != null)
          processorListener.afterFailure(containerException);
        else
          processorListener.afterStop();
      }
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