@Override
    public void afterFailure(Throwable t) {
      // we need to close associated coordinator metadata store, although the processor failed
      closeAndRemoveProcessor();

      // the processor stopped with failure, this is logging the first processor's failure as the cause of
      // the whole application failure
      if (failure.compareAndSet(null, t)) {
        // shutdown the other processors
        processors.forEach(sp -> {
          sp.getLeft().stop();     // Stop StreamProcessor
          if (sp.getRight() != null) {
            sp.getRight().close(); // Close associated coordinator metadata store
          }
        });
      }

      // handle the current processor's shutdown failure.
      handleProcessorShutdown(t);
    }
Defines a specific implementation of {@link ProcessorLifecycleListener} for local {@link StreamProcessor}s.
samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java