LocalApplicationRunner implements ApplicationRunner { @VisibleForTesting static String createProcessorId(ApplicationConfig appConfig) { if (StringUtils.isNotBlank(appConfig.getProcessorId())) { return appConfig.getProcessorId(); } else if (StringUtils.isNotBlank(appConfig.getAppProcessorIdGeneratorClass())) { ProcessorIdGenerator idGenerator = ReflectionUtil.getObj(appConfig.getAppProcessorIdGeneratorClass(), ProcessorIdGenerator.class); return idGenerator.generateProcessorId(appConfig); } else { throw new ConfigException(String.format("Expected either %s or %s to be configured", ApplicationConfig.PROCESSOR_ID, ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS)); } } 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 testCreateProcessorIdShouldReturnProcessorIdDefinedInConfiguration() { String processorId = "testProcessorId"; MapConfig configMap = new MapConfig(ImmutableMap.of(ApplicationConfig.PROCESSOR_ID, processorId)); String actualProcessorId = LocalApplicationRunner.createProcessorId(new ApplicationConfig(configMap)); assertEquals(processorId, actualProcessorId); }
@Test public void testCreateProcessorIdShouldInvokeProcessorIdGeneratorDefinedInConfiguration() { String processorId = "testProcessorId"; MapConfig configMap = new MapConfig(ImmutableMap.of(ApplicationConfig.APP_PROCESSOR_ID_GENERATOR_CLASS, MockProcessorIdGenerator.class.getCanonicalName())); String actualProcessorId = LocalApplicationRunner.createProcessorId(new ApplicationConfig(configMap)); assertEquals(processorId, actualProcessorId); }
@Test(expected = ConfigException.class) public void testCreateProcessorIdShouldThrowExceptionWhenProcessorIdAndGeneratorAreNotDefined() { ApplicationConfig mockConfig = Mockito.mock(ApplicationConfig.class); Mockito.when(mockConfig.getProcessorId()).thenReturn(null); LocalApplicationRunner.createProcessorId(mockConfig); }