diff --git a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java index 5549570397..9337d9bd22 100644 --- a/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java +++ b/samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java @@ -83,7 +83,7 @@ public class LocalApplicationRunner implements ApplicationRunner { private final AtomicReference failure = new AtomicReference<>(); private final boolean isAppModeBatch; private final Optional coordinationUtils; - private final MetadataStoreFactory metadataStoreFactory; + private final Optional metadataStoreFactory; private Optional runId = Optional.empty(); private Optional runIdGenerator = Optional.empty(); @@ -96,7 +96,7 @@ public class LocalApplicationRunner implements ApplicationRunner { * @param config configuration for the application */ public LocalApplicationRunner(SamzaApplication app, Config config) { - this(app, config, new CoordinatorStreamMetadataStoreFactory()); + this(app, config, getMetadataStoreFactory(new JobConfig(config))); } /** @@ -110,7 +110,7 @@ public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStore this.appDesc = ApplicationDescriptorUtil.getAppDescriptor(app, config); this.isAppModeBatch = new ApplicationConfig(config).getAppMode() == ApplicationConfig.ApplicationMode.BATCH; this.coordinationUtils = getCoordinationUtils(config, getClass().getClassLoader()); - this.metadataStoreFactory = metadataStoreFactory; + this.metadataStoreFactory = Optional.ofNullable(metadataStoreFactory); } /** @@ -121,7 +121,16 @@ public LocalApplicationRunner(SamzaApplication app, Config config, MetadataStore this.appDesc = appDesc; this.isAppModeBatch = new ApplicationConfig(appDesc.getConfig()).getAppMode() == ApplicationConfig.ApplicationMode.BATCH; this.coordinationUtils = coordinationUtils; - this.metadataStoreFactory = new CoordinatorStreamMetadataStoreFactory(); + this.metadataStoreFactory = Optional.ofNullable(getMetadataStoreFactory(new JobConfig(appDesc.getConfig()))); + } + + static MetadataStoreFactory getMetadataStoreFactory(JobConfig jobConfig) { + if (jobConfig.getCoordinatorSystemNameOrNull() != null) { + return new CoordinatorStreamMetadataStoreFactory(); + } + LOG.warn("{} or {} not configured. No coordinator stream metadata store will be created.", + JobConfig.JOB_COORDINATOR_SYSTEM, JobConfig.JOB_DEFAULT_SYSTEM); + return null; } private Optional getCoordinationUtils(Config config, ClassLoader classLoader) { @@ -189,7 +198,9 @@ public void run(ExternalContext externalContext) { jobConfigs.forEach(jobConfig -> { LOG.debug("Starting job {} StreamProcessor with config {}", jobConfig.getName(), jobConfig); MetadataStore coordinatorStreamStore = createCoordinatorStreamStore(jobConfig); - coordinatorStreamStore.init(); + if (coordinatorStreamStore != null) { + coordinatorStreamStore.init(); + } StreamProcessor processor = createStreamProcessor(jobConfig, appDesc, sp -> new LocalStreamProcessorLifecycleListener(sp, jobConfig), Optional.ofNullable(externalContext), coordinatorStreamStore); processors.add(Pair.of(processor, coordinatorStreamStore)); @@ -262,9 +273,12 @@ CountDownLatch getShutdownLatch() { @VisibleForTesting MetadataStore createCoordinatorStreamStore(Config jobConfig) { - MetadataStore coordinatorStreamStore = - metadataStoreFactory.getMetadataStore("NoOp", jobConfig, new MetricsRegistryMap()); - return coordinatorStreamStore; + if (metadataStoreFactory.isPresent()) { + MetadataStore coordinatorStreamStore = + metadataStoreFactory.get().getMetadataStore("NoOp", jobConfig, new MetricsRegistryMap()); + return coordinatorStreamStore; + } + return null; } @VisibleForTesting diff --git a/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java b/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java index 9e63cc47d8..4acc8c3791 100644 --- a/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java +++ b/samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java @@ -44,6 +44,7 @@ import org.apache.samza.coordinator.metadatastore.CoordinatorStreamStore; import org.apache.samza.execution.LocalJobPlanner; import org.apache.samza.job.ApplicationStatus; +import org.apache.samza.metadatastore.MetadataStoreFactory; import org.apache.samza.processor.StreamProcessor; import org.apache.samza.task.IdentityStreamTask; import org.apache.samza.zk.ZkMetadataStore; @@ -60,6 +61,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyObject; @@ -210,6 +212,43 @@ public void testRunComplete() throws Exception { assertEquals(runner.status(), ApplicationStatus.SuccessfulFinish); } + @Test + public void testRunCompleteWithouCoordinatorStreamStore() throws Exception { + Map cfgs = new HashMap<>(); + cfgs.put(ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS, UUIDGenerator.class.getName()); + config = new MapConfig(cfgs); + ProcessorLifecycleListenerFactory mockFactory = (pContext, cfg) -> mock(ProcessorLifecycleListener.class); + mockApp = (StreamApplication) appDesc -> { + appDesc.withProcessorLifecycleListenerFactory(mockFactory); + }; + prepareTest(); + + // return the jobConfigs from the planner + doReturn(Collections.singletonList(new JobConfig(new MapConfig(config)))).when(localPlanner).prepareJobs(); + + StreamProcessor sp = mock(StreamProcessor.class); + ArgumentCaptor captor = + ArgumentCaptor.forClass(StreamProcessor.StreamProcessorLifecycleListenerFactory.class); + + doAnswer(i -> + { + ProcessorLifecycleListener listener = captor.getValue().createInstance(sp); + listener.afterStart(); + listener.afterStop(); + return null; + }).when(sp).start(); + + ExternalContext externalContext = mock(ExternalContext.class); + doReturn(sp).when(runner) + .createStreamProcessor(anyObject(), anyObject(), captor.capture(), eq(Optional.of(externalContext)), eq(null)); + doReturn(null).when(runner).createCoordinatorStreamStore(any(Config.class)); + + runner.run(externalContext); + runner.waitForFinish(); + + assertEquals(runner.status(), ApplicationStatus.SuccessfulFinish); + } + @Test public void testRunFailure() throws Exception { Map cfgs = new HashMap<>(); @@ -437,4 +476,24 @@ private void prepareTestForRunId() throws Exception { doReturn(coordinatorStreamStore).when(runner).createCoordinatorStreamStore(any(Config.class)); } + @Test + public void testGetMetadataStoreFactoryWithoutJobCoordinatorSystem() { + MetadataStoreFactory metadataStoreFactory = + LocalApplicationRunner.getMetadataStoreFactory(new JobConfig(new MapConfig())); + assertNull(metadataStoreFactory); + } + + @Test + public void testGetMetadataStoreFactoryWithJobCoordinatorSystem() { + MetadataStoreFactory metadataStoreFactory = + LocalApplicationRunner.getMetadataStoreFactory(new JobConfig(new MapConfig(ImmutableMap.of(JobConfig.JOB_COORDINATOR_SYSTEM, "test-system")))); + assertNotNull(metadataStoreFactory); + } + + @Test + public void testGetMetadataStoreFactoryWithDefaultSystem() { + MetadataStoreFactory metadataStoreFactory = + LocalApplicationRunner.getMetadataStoreFactory(new JobConfig(new MapConfig(ImmutableMap.of(JobConfig.JOB_DEFAULT_SYSTEM, "test-system")))); + assertNotNull(metadataStoreFactory); + } }