ConfigUtil { public static Config loadConfig(Config original) { JobConfig jobConfig = new JobConfig(original); if (!jobConfig.getConfigLoaderFactory().isPresent()) { throw new ConfigException("Missing key " + JobConfig.CONFIG_LOADER_FACTORY + "."); } ConfigLoaderFactory factory = ReflectionUtil.getObj(jobConfig.getConfigLoaderFactory().get(), ConfigLoaderFactory.class); ConfigLoader loader = factory.getLoader(original.subset(ConfigLoaderFactory.CONFIG_LOADER_PROPERTIES_PREFIX)); return rewriteConfig(override(loader.getConfig(), original)); }  static Config rewriteConfig(Config config); static Config applyRewriter(Config config, String rewriterName); static Config loadConfig(Config original);  }
@Test(expected = ConfigException.class) public void testLoadConfigWithoutLoader() { Map<String, String> config = new HashMap<>(); config.put(JobConfig.JOB_NAME, "new-test-job"); ConfigUtil.loadConfig(new MapConfig(config)); }
@Test public void testLoadConfigWithOverridesAndRewrites() { Map<String, String> config = new HashMap<>(); config.put(JobConfig.CONFIG_LOADER_FACTORY, PropertiesConfigLoaderFactory.class.getCanonicalName()); config.put(JobConfig.JOB_NAME, "new-test-job"); config.put(JobConfig.CONFIG_REWRITERS, REWRITER_NAME); config.put(CONFIG_KEY, CONFIG_VALUE); config.put(String.format(JobConfig.CONFIG_REWRITER_CLASS, REWRITER_NAME), NewPropertyRewriter.class.getName()); config.put(PropertiesConfigLoaderFactory.CONFIG_LOADER_PROPERTIES_PREFIX + "path", getClass().getResource("/test.properties").getPath()); Config actual = ConfigUtil.loadConfig(new MapConfig(config)); assertEquals("org.apache.samza.job.MockJobFactory", actual.get("job.factory.class")); assertEquals("bar", actual.get("foo")); assertEquals("new-test-job", actual.get("job.name")); assertEquals(CONFIG_VALUE, actual.get(NEW_CONFIG_KEY)); }