ConfigUtil { public static Config rewriteConfig(Config config) { Optional<String> configRewriterNamesOptional = new JobConfig(config).getConfigRewriters(); if (configRewriterNamesOptional.isPresent()) { String[] configRewriterNames = configRewriterNamesOptional.get().split(","); Config rewrittenConfig = config; for (String configRewriterName : configRewriterNames) { rewrittenConfig = applyRewriter(rewrittenConfig, configRewriterName); } return rewrittenConfig; } else { return config; } }  static Config rewriteConfig(Config config); static Config applyRewriter(Config config, String rewriterName); static Config loadConfig(Config original);  }
@Test public void testRewriteConfig() { Map<String, String> baseConfigMap = ImmutableMap.of(CONFIG_KEY, CONFIG_VALUE); Map<String, String> fullConfig = new HashMap<>(baseConfigMap); assertEquals(fullConfig, ConfigUtil.rewriteConfig(new MapConfig(fullConfig))); fullConfig = new HashMap<>(baseConfigMap); fullConfig.put(JobConfig.CONFIG_REWRITERS, REWRITER_NAME); fullConfig.put(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.rewriteConfig(new MapConfig(fullConfig))); fullConfig = new HashMap<>(baseConfigMap); fullConfig.put(JobConfig.CONFIG_REWRITERS, REWRITER_NAME); fullConfig.put(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.rewriteConfig(new MapConfig(fullConfig))); fullConfig = new HashMap<>(baseConfigMap); fullConfig.put(JobConfig.CONFIG_REWRITERS, REWRITER_NAME); fullConfig.put(String.format(JobConfig.CONFIG_REWRITER_CLASS, REWRITER_NAME), DeletePropertyRewriter.class.getName()); expectedConfigMap = new HashMap<>(fullConfig); expectedConfigMap.remove(CONFIG_KEY); assertEquals(new MapConfig(expectedConfigMap), ConfigUtil.rewriteConfig(new MapConfig(fullConfig))); fullConfig = new HashMap<>(baseConfigMap); fullConfig.put(JobConfig.CONFIG_REWRITERS, OTHER_REWRITER_NAME); fullConfig.put(String.format(JobConfig.CONFIG_REWRITER_CLASS, REWRITER_NAME), NewPropertyRewriter.class.getName()); fullConfig.put(String.format(JobConfig.CONFIG_REWRITER_CLASS, OTHER_REWRITER_NAME), UpdatePropertyRewriter.class.getName()); expectedConfigMap = new HashMap<>(fullConfig); expectedConfigMap.put(CONFIG_KEY, CONFIG_VALUE + CONFIG_VALUE); assertEquals(new MapConfig(expectedConfigMap), ConfigUtil.rewriteConfig(new MapConfig(fullConfig))); fullConfig = new HashMap<>(baseConfigMap); fullConfig.put(JobConfig.CONFIG_REWRITERS, REWRITER_NAME + "," + OTHER_REWRITER_NAME); fullConfig.put(String.format(JobConfig.CONFIG_REWRITER_CLASS, REWRITER_NAME), NewPropertyRewriter.class.getName()); fullConfig.put(String.format(JobConfig.CONFIG_REWRITER_CLASS, OTHER_REWRITER_NAME), UpdatePropertyRewriter.class.getName()); expectedConfigMap = new HashMap<>(fullConfig); expectedConfigMap.put(NEW_CONFIG_KEY, CONFIG_VALUE + CONFIG_VALUE); assertEquals(new MapConfig(expectedConfigMap), ConfigUtil.rewriteConfig(new MapConfig(fullConfig))); }
@Test(expected = SamzaException.class) public void testRewriteConfigConfigRewritersEmptyString() { Config config = new MapConfig(ImmutableMap.of(JobConfig.CONFIG_REWRITERS, "")); ConfigUtil.rewriteConfig(config); }
@Test(expected = SamzaException.class) public void testRewriteConfigNoClassForConfigRewriterName() { Config config = new MapConfig(ImmutableMap.of(CONFIG_KEY, CONFIG_VALUE, JobConfig.CONFIG_REWRITERS, "unknownRewriter")); ConfigUtil.rewriteConfig(config); }
@Test(expected = SamzaException.class) public void testRewriteConfigRewriterClassDoesNotExist() { Config config = new MapConfig(ImmutableMap.of(CONFIG_KEY, CONFIG_VALUE, JobConfig.CONFIG_REWRITERS, REWRITER_NAME, String.format(JobConfig.CONFIG_REWRITER_CLASS, REWRITER_NAME), "not_a_class")); ConfigUtil.rewriteConfig(config); }