ConfigUtil { public static Config applyRewriter(Config config, String rewriterName) { String rewriterClassName = new JobConfig(config).getConfigRewriterClass(rewriterName) .orElseThrow(() -> new SamzaException( String.format("Unable to find class config for config rewriter %s.", rewriterName))); ConfigRewriter rewriter = ReflectionUtil.getObj(rewriterClassName, ConfigRewriter.class); LOG.info("Re-writing config with {}", rewriter); return rewriter.rewrite(rewriterName, config); }  static Config rewriteConfig(Config config); static Config applyRewriter(Config config, String rewriterName); static Config loadConfig(Config original);  }
@Test public void testApplyRewriter() { Map<String, String> fullConfig = ImmutableMap.of(CONFIG_KEY, CONFIG_VALUE, String.format(JobConfig.CONFIG_REWRITER_CLASS, REWRITER_NAME), NewPropertyRewriter.class.getName()); Map<String, String> expectedConfigMap = new HashMap<>(fullConfig); expectedConfigMap.put(NEW_CONFIG_KEY, CONFIG_VALUE); assertEquals(new MapConfig(expectedConfigMap), ConfigUtil.applyRewriter(new MapConfig(fullConfig), REWRITER_NAME)); fullConfig = ImmutableMap.of(CONFIG_KEY, CONFIG_VALUE, String.format(JobConfig.CONFIG_REWRITER_CLASS, REWRITER_NAME), UpdatePropertyRewriter.class.getName()); expectedConfigMap = new HashMap<>(fullConfig); expectedConfigMap.put(CONFIG_KEY, CONFIG_VALUE + CONFIG_VALUE); assertEquals(new MapConfig(expectedConfigMap), ConfigUtil.applyRewriter(new MapConfig(fullConfig), REWRITER_NAME)); fullConfig = ImmutableMap.of(CONFIG_KEY, CONFIG_VALUE, String.format(JobConfig.CONFIG_REWRITER_CLASS, REWRITER_NAME), DeletePropertyRewriter.class.getName()); expectedConfigMap = new HashMap<>(fullConfig); expectedConfigMap.remove(CONFIG_KEY); assertEquals(new MapConfig(expectedConfigMap), ConfigUtil.applyRewriter(new MapConfig(fullConfig), REWRITER_NAME)); }
@Test(expected = SamzaException.class) public void testApplyRewriterNoClassForConfigRewriterName() { Map<String, String> fullConfig = ImmutableMap.of(CONFIG_KEY, CONFIG_VALUE); ConfigUtil.applyRewriter(new MapConfig(fullConfig), REWRITER_NAME); }
@Test(expected = SamzaException.class) public void testApplyRewriterClassDoesNotExist() { Map<String, String> fullConfig = ImmutableMap.of(CONFIG_KEY, CONFIG_VALUE, String.format(JobConfig.CONFIG_REWRITER_CLASS, REWRITER_NAME), "not_a_class"); Config expectedConfig = new MapConfig(ImmutableMap.of(CONFIG_KEY, CONFIG_VALUE, NEW_CONFIG_KEY, CONFIG_VALUE)); assertEquals(expectedConfig, ConfigUtil.applyRewriter(new MapConfig(fullConfig), REWRITER_NAME)); }