LocalApplicationRunner implements ApplicationRunner { @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(); 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(); } LocalApplicationRunner(SamzaApplication app, Config config);  LocalApplicationRunner(SamzaApplication app, Config config, MetadataStoreFactory metadataStoreFactory); @VisibleForTesting  LocalApplicationRunner(SamzaApplication app, Config config, CoordinationUtils coordinationUtils); private  LocalApplicationRunner(LocalApplicationRunnerContext context); Optional<String> getRunId(); @Override void run(ExternalContext externalContext); @Override void kill(); @Override ApplicationStatus status(); @Override void waitForFinish(); @Override boolean waitForFinish(Duration timeout);  }
@Test public void testGetCoordinatorStreamStoreFactoryWithoutJobCoordinatorSystem() { Optional<MetadataStoreFactory> metadataStoreFactory = LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(new MapConfig()); assertFalse(metadataStoreFactory.isPresent()); }
@Test public void testGetCoordinatorStreamStoreFactoryWithJobCoordinatorSystem() { Optional<MetadataStoreFactory> metadataStoreFactory = LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(new MapConfig(ImmutableMap.of(JobConfig.JOB_COORDINATOR_SYSTEM, "test-system"))); assertTrue(metadataStoreFactory.isPresent()); }
@Test public void testGetCoordinatorStreamStoreFactoryWithDefaultSystem() { Optional<MetadataStoreFactory> metadataStoreFactory = LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(new MapConfig(ImmutableMap.of(JobConfig.JOB_DEFAULT_SYSTEM, "test-system"))); assertTrue(metadataStoreFactory.isPresent()); }
@Test public void testGetCoordinatorStreamStoreFactoryWithNonZkJobCoordinator() { MapConfig mapConfig = new MapConfig( ImmutableMap.of( JobConfig.JOB_DEFAULT_SYSTEM, "test-system", JobCoordinatorConfig.JOB_COORDINATOR_FACTORY, PassthroughJobCoordinatorFactory.class.getName())); Optional<MetadataStoreFactory> metadataStoreFactory = LocalApplicationRunner.getDefaultCoordinatorStreamStoreFactory(mapConfig); assertFalse(metadataStoreFactory.isPresent()); }