private Optional<StreamRegexMonitor> getInputRegexMonitor(JobModel jobModel, Config config, SystemAdmins systemAdmins) {
    StreamMetadataCache streamMetadataCache = new StreamMetadataCache(systemAdmins, 0, SystemClock.instance());
    return new StreamRegexMonitorFactory(streamMetadataCache, this.metrics).build(jobModel, config,
      (initialInputSet, newInputStreams, regexesMonitored) -> {
        if (hasDurableStores) {
          LOG.error("New input system-streams discovered. Failing the job. New input streams: {}"
              + " Existing input streams: {}", newInputStreams, initialInputSet);
          state.status = SamzaApplicationState.SamzaAppStatus.FAILED;
        }
        coordinatorCallbackException = new InputStreamsDiscoveredException("New input streams discovered: " + newInputStreams);
      });
  }
Stops all components of the JobCoordinator.
samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinator.java