@Override
  public void start() {
    LOG.info("Starting job coordinator");
    this.systemAdmins.start();
    this.startpointManager.ifPresent(StartpointManager::start);
    try {
      JobModel jobModel = newJobModel();
      doSetLoggingContextConfig(jobModel.getConfig());
      // monitors should be created right after job model is calculated (see jobModelMonitors() for more details)
      JobModelMonitors jobModelMonitors = jobModelMonitors(jobModel);
      Optional<DiagnosticsManager> diagnosticsManager = diagnosticsManager(jobModel);
      JobCoordinatorMetadata newMetadata =
          this.jobCoordinatorMetadataManager.generateJobCoordinatorMetadata(jobModel, jobModel.getConfig());
      Set<JobMetadataChange> jobMetadataChanges = checkForMetadataChanges(newMetadata);
      if (!jobMetadataChanges.isEmpty() && !jobMetadataChanges.contains(JobMetadataChange.NEW_DEPLOYMENT)) {
        /*
         * If the job coordinator comes up, but not due to a new deployment, and the metadata changed, then trigger a
         * restart. This case applies if the job coordinator died and the job model needed to change while it was down.
         * If there were no metadata changes, then just let the current workers continue to run.
         * If there was a new deployment (which includes the case where the coordinator requested a restart), then we
         * rely on the external resource manager to make sure the previous workers restarted, so we don't need to
         * restart again.
         */
        LOG.info("Triggering job restart");
        this.jobRestartSignal.restartJob();
      } else {
        prepareWorkerExecution(jobModel, newMetadata, jobMetadataChanges);
        // save components that depend on job model in order to manage lifecycle or access later
        this.currentDiagnosticsManager = diagnosticsManager;
        this.currentJobModelMonitors = Optional.of(jobModelMonitors);
        this.currentJobModel = Optional.of(jobModel);
        // lifecycle: start components
        this.coordinatorCommunication.start();
        this.jobCoordinatorListener.ifPresent(listener -> listener.onNewJobModel(this.processorId, jobModel));
        this.currentDiagnosticsManager.ifPresent(DiagnosticsManager::start);
        jobModelMonitors.start();
        this.jobPreparationComplete.set(true);
      }
    } catch (Exception e) {
      LOG.error("Error while running job coordinator; exiting", e);
      throw new SamzaException("Error while running job coordinator", e);
    }
  }
Keeps track of if the job coordinator has completed all preparation for running the job, including
publishing a new job model and starting the job model monitors.
samza-core/src/main/java/org/apache/samza/coordinator/staticresource/StaticResourceJobCoordinator.java