@VisibleForTesting
  SamzaContainer createSamzaContainer(String processorId, JobModel jobModel) {
    // Creating diagnostics manager and wiring it respectively
    String jobName = new JobConfig(config).getName().get();
    String jobId = new JobConfig(config).getJobId();
    Optional<DiagnosticsManager> diagnosticsManager =
        DiagnosticsUtil.buildDiagnosticsManager(jobName, jobId, jobModel, processorId, Optional.empty(),
            Optional.empty(), config);

    // Metadata store lifecycle managed outside of the SamzaContainer.
    // All manager lifecycles are managed in the SamzaContainer including startpointManager
    StartpointManager startpointManager = null;
    if (metadataStore != null && new JobConfig(config).getStartpointEnabled()) {
      startpointManager = new StartpointManager(metadataStore);
    } else if (!new JobConfig(config).getStartpointEnabled()) {
      LOGGER.warn("StartpointManager not instantiated because startpoints is not enabled");
    } else {
      LOGGER.warn("StartpointManager cannot be instantiated because no metadata store defined for this stream processor");
    }
    /*
     * StreamProcessor has a metricsRegistry instance variable, but StreamProcessor registers its metrics on its own
     * with the reporters. Therefore, don't reuse the StreamProcessor.metricsRegistry, because SamzaContainer also
     * registers the registry, and that will result in unnecessary duplicate metrics.
     */
    MetricsRegistryMap metricsRegistryMap = new MetricsRegistryMap();

    DrainMonitor drainMonitor = null;
    JobConfig jobConfig = new JobConfig(config);
    if (metadataStore != null && jobConfig.getDrainMonitorEnabled()) {
      drainMonitor = new DrainMonitor(metadataStore, config);
    }

    return SamzaContainer.apply(processorId, jobModel, ScalaJavaUtil.toScalaMap(this.customMetricsReporter),
        metricsRegistryMap, this.taskFactory, JobContextImpl.fromConfigWithDefaults(this.config, jobModel),
        Option.apply(this.applicationDefinedContainerContextFactoryOptional.orElse(null)),
        Option.apply(this.applicationDefinedTaskContextFactoryOptional.orElse(null)),
        Option.apply(this.externalContextOptional.orElse(null)), null, startpointManager,
        Option.apply(diagnosticsManager.orElse(null)), drainMonitor);
  }
<p>
Asynchronously stops the {@link StreamProcessor}'s running components - {@link SamzaContainer}
and {@link JobCoordinator}
</p>
Here're the ways which can stop the StreamProcessor:
<ol>
  <li>Caller of StreamProcessor invokes stop()</li>
  <li>Samza Container completes processing (eg. bounded input) and shuts down</li>
  <li>Samza Container fails</li>
  <li>Job Coordinator fails</li>
</ol>
When either container or coordinator stops (cleanly or due to exception), it will try to shutdown the
StreamProcessor. This needs to be synchronized so that only one code path gets triggered for shutdown.
<br>
If container is running,
<ol>
  <li>container is shutdown cleanly and {@link SamzaContainerListener#afterStop()} will trigger
  {@link JobCoordinator#stop()}</li>
  <li>container fails to shutdown cleanly and {@link SamzaContainerListener#afterFailure(Throwable)} will
  trigger {@link JobCoordinator#stop()}</li>
</ol>
If container is not running, then this method will simply shutdown the {@link JobCoordinator}.
samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java