private static void runJobCoordinator(String jobCoordinatorClassName, MetricsRegistryMap metrics,
      MetadataStore metadataStore, Config finalConfig) {
    JobCoordinatorFactory jobCoordinatorFactory =
        ReflectionUtil.getObj(jobCoordinatorClassName, JobCoordinatorFactory.class);
    JobCoordinator jobCoordinator =
        jobCoordinatorFactory.getJobCoordinator(JOB_COORDINATOR_PROCESSOR_ID_PLACEHOLDER, finalConfig, metrics,
            metadataStore);
    Map<String, MetricsReporter> metricsReporters =
        MetricsReporterLoader.getMetricsReporters(new MetricsConfig(finalConfig),
            CoordinationConstants.JOB_COORDINATOR_CONTAINER_NAME);
    metricsReporters.values()
        .forEach(
          metricsReporter -> metricsReporter.register(CoordinationConstants.JOB_COORDINATOR_CONTAINER_NAME, metrics));
    metricsReporters.values().forEach(MetricsReporter::start);
    CountDownLatch waitForShutdownLatch = new CountDownLatch(1);
    jobCoordinator.setListener(new NoProcessorJobCoordinatorListener(waitForShutdownLatch));
    jobCoordinator.start();
    addShutdownHook(jobCoordinator);
    try {
      waitForShutdownLatch.await();
    } catch (InterruptedException e) {
      String errorMessage = "Error while waiting for coordinator to complete";
      LOG.error(errorMessage, e);
      throw new SamzaException(errorMessage, e);
    } finally {
      metricsReporters.values().forEach(MetricsReporter::stop);
    }
  }
Run {@link ClusterBasedJobCoordinator} with full job config.

@param app SamzaApplication to run.
@param config full job config.
samza-core/src/main/java/org/apache/samza/clustermanager/JobCoordinatorLaunchUtil.java