LocalApplicationRunner implements ApplicationRunner { @Override public void run(ExternalContext externalContext) { initializeRunId(); LocalJobPlanner planner = getPlanner(); try { List<JobConfig> jobConfigs = planner.prepareJobs(); if (jobConfigs.isEmpty()) { throw new SamzaException("No jobs to run."); } jobConfigs.forEach(jobConfig -> { LOG.debug("Starting job {} StreamProcessor with config {}", jobConfig.getName(), jobConfig); MetadataStore coordinatorStreamStore = createCoordinatorStreamStore(jobConfig); 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)); }); numProcessorsToStart.set(processors.size()); processors.forEach(sp -> sp.getLeft().start()); } catch (Throwable throwable) { cleanup(); appStatus = ApplicationStatus.unsuccessfulFinish(throwable); shutdownLatch.countDown(); throw new SamzaException(String.format("Failed to start application: %s", new ApplicationConfig(appDesc.getConfig()).getGlobalAppId()), throwable); } } 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 testRunIdForBatch() throws Exception { final Map<String, String> cfgs = new HashMap<>(); cfgs.put(ApplicationConfig.APP_MODE, "BATCH"); cfgs.put(ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS, UUIDGenerator.class.getName()); cfgs.put(JobConfig.JOB_NAME, "test-task-job"); cfgs.put(JobConfig.JOB_ID, "jobId"); config = new MapConfig(cfgs); mockApp = new LegacyTaskApplication(IdentityStreamTask.class.getName()); prepareTestForRunId(); runner.run(); verify(coordinationUtils, Mockito.times(1)).getLock(CoordinationConstants.RUNID_LOCK_ID); verify(clusterMembership, Mockito.times(1)).getNumberOfProcessors(); verify(metadataStore, Mockito.times(1)).put(eq(CoordinationConstants.RUNID_STORE_KEY), any(byte[].class)); verify(metadataStore, Mockito.times(1)).flush(); }
@Test public void testRunIdForStream() throws Exception { final Map<String, String> cfgs = new HashMap<>(); cfgs.put(ApplicationConfig.APP_MODE, "STREAM"); cfgs.put(ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS, UUIDGenerator.class.getName()); cfgs.put(JobConfig.JOB_NAME, "test-task-job"); cfgs.put(JobConfig.JOB_ID, "jobId"); config = new MapConfig(cfgs); mockApp = new LegacyTaskApplication(IdentityStreamTask.class.getName()); prepareTestForRunId(); runner.run(); verify(coordinationUtils, Mockito.times(0)).getLock(CoordinationConstants.RUNID_LOCK_ID); verify(coordinationUtils, Mockito.times(0)).getClusterMembership(); verify(clusterMembership, Mockito.times(0)).getNumberOfProcessors(); verify(metadataStore, Mockito.times(0)).put(eq(CoordinationConstants.RUNID_STORE_KEY), any(byte[].class)); verify(metadataStore, Mockito.times(0)).flush(); }