private StreamPartitionCountMonitor getPartitionCountMonitor(Config config, SystemAdmins systemAdmins) {
    StreamMetadataCache streamMetadataCache = new StreamMetadataCache(systemAdmins, 0, SystemClock.instance());
    return new StreamPartitionCountMonitorFactory(streamMetadataCache, this.metrics).build(config, streamsChanged -> {
      // Fail the jobs with durable state store. Otherwise, application state.status remains UNDEFINED s.t. YARN job will be restarted
      if (hasDurableStores) {
        LOG.error(
            "Input topic partition count changed in a job with durable state. Failing the job. Changed topics: {}",
            streamsChanged.toString());
        state.status = SamzaApplicationState.SamzaAppStatus.FAILED;
      }
      coordinatorCallbackException = new PartitionChangeException(
          "Input topic partition count changes detected for topics: " + streamsChanged.toString());
    });
  }
Stops all components of the JobCoordinator.
samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinator.java