ApplicationUtil { public static SamzaApplication fromConfig(Config config) { String appClassName = new ApplicationConfig(config).getAppClass(); if (StringUtils.isNotBlank(appClassName)) { try { Class<SamzaApplication> appClass = (Class<SamzaApplication>) Class.forName(appClassName); if (StreamApplication.class.isAssignableFrom(appClass) || TaskApplication.class.isAssignableFrom(appClass)) { return appClass.newInstance(); } } catch (IllegalAccessException | InstantiationException | ClassNotFoundException e) { throw new ConfigException(String.format("Loading app.class %s failed. The user application has to implement " + "StreamApplication or TaskApplication.", appClassName), e); } } TaskConfig taskConfig = new TaskConfig(config); Optional<String> taskClassOption = taskConfig.getTaskClass(); if (!taskClassOption.isPresent() || StringUtils.isBlank(taskClassOption.get())) { throw new ConfigException("Legacy task applications must set a non-empty task.class in configuration."); } return new LegacyTaskApplication(taskClassOption.get()); }  static SamzaApplication fromConfig(Config config);  }
@Test public void testStreamAppClass() { Map<String, String> configMap = new HashMap<>(); configMap.put(ApplicationConfig.APP_CLASS, MockStreamApplication.class.getName()); SamzaApplication app = ApplicationUtil.fromConfig(new MapConfig(configMap)); assertTrue(app instanceof MockStreamApplication); configMap.put(TaskConfig.TASK_CLASS, MockStreamTask.class.getName()); app = ApplicationUtil.fromConfig(new MapConfig(configMap)); assertTrue(app instanceof MockStreamApplication); }
@Test public void testTaskAppClass() { Map<String, String> configMap = new HashMap<>(); configMap.put(ApplicationConfig.APP_CLASS, MockTaskApplication.class.getName()); SamzaApplication app = ApplicationUtil.fromConfig(new MapConfig(configMap)); assertTrue(app instanceof MockTaskApplication); configMap.put(TaskConfig.TASK_CLASS, MockStreamTask.class.getName()); app = ApplicationUtil.fromConfig(new MapConfig(configMap)); assertTrue(app instanceof MockTaskApplication); }
@Test public void testTaskClassOnly() { Map<String, String> configMap = new HashMap<>(); configMap.put(TaskConfig.TASK_CLASS, MockStreamTask.class.getName()); Config config = new MapConfig(configMap); SamzaApplication app = ApplicationUtil.fromConfig(config); assertTrue(app instanceof TaskApplication); TaskApplicationDescriptorImpl appSpec = new TaskApplicationDescriptorImpl((TaskApplication) app, config); assertTrue(appSpec.getTaskFactory().createInstance() instanceof MockStreamTask); }
@Test(expected = ConfigException.class) public void testEmptyTaskClassOnly() { Map<String, String> configMap = new HashMap<>(); configMap.put(TaskConfig.TASK_CLASS, ""); ApplicationUtil.fromConfig(new MapConfig(configMap)); }
@Test(expected = ConfigException.class) public void testNoAppClassNoTaskClass() { Map<String, String> configMap = new HashMap<>(); ApplicationUtil.fromConfig(new MapConfig(configMap)); }