@VisibleForTesting
  static Optional<MetadataStoreFactory> getDefaultCoordinatorStreamStoreFactory(Config config) {
    JobConfig jobConfig = new JobConfig(config);

    String coordinatorSystemName = jobConfig.getCoordinatorSystemNameOrNull();
    JobCoordinatorConfig jobCoordinatorConfig = new JobCoordinatorConfig(jobConfig);
    String jobCoordinatorFactoryClassName = jobCoordinatorConfig.getJobCoordinatorFactoryClassName();

    // TODO: Remove restriction to only ZkJobCoordinator after next phase of metadata store abstraction.
    if (StringUtils.isNotBlank(coordinatorSystemName) && ZkJobCoordinatorFactory.class.getName().equals(jobCoordinatorFactoryClassName)) {
      return Optional.of(new CoordinatorStreamMetadataStoreFactory());
    }

    LOG.warn("{} or {} not configured, or {} is not {}. No default coordinator stream metadata store will be created.",
        JobConfig.JOB_COORDINATOR_SYSTEM, JobConfig.JOB_DEFAULT_SYSTEM,
        JobCoordinatorConfig.JOB_COORDINATOR_FACTORY, ZkJobCoordinatorFactory.class.getName());
    return Optional.empty();
  }
Constructor only used in unit test to allow injection of {@link LocalJobPlanner}
samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java