ConfigurableIngestTopology { protected void configureStorm(Configuration conf, Config stormConf) throws IllegalAccessException { stormConf.registerSerialization(LogRecord.class); stormConf.registerMetricsConsumer(LoggingMetricsConsumer.class); for (Iterator<String> iter = conf.getKeys(); iter.hasNext(); ) { String key = iter.next(); String keyString = key.toString(); String cleanedKey = keyString.replaceAll("\\.\\.", "."); String schemaFieldName = cleanedKey.replaceAll("\\.", "_").toUpperCase() + "_SCHEMA"; Field field = FieldUtils.getField(Config.class, schemaFieldName); Object fieldObject = field.get(null); if (fieldObject == Boolean.class) stormConf.put(cleanedKey, conf.getBoolean(keyString)); else if (fieldObject == String.class) stormConf.put(cleanedKey, conf.getString(keyString)); else if (fieldObject == ConfigValidation.DoubleValidator) stormConf.put(cleanedKey, conf.getDouble(keyString)); else if (fieldObject == ConfigValidation.IntegerValidator) stormConf.put(cleanedKey, conf.getInt(keyString)); else if (fieldObject == ConfigValidation.PowerOf2Validator) stormConf.put(cleanedKey, conf.getLong(keyString)); else if (fieldObject == ConfigValidation.StringOrStringListValidator) stormConf.put(cleanedKey, Arrays.asList(conf.getStringArray(keyString))); else if (fieldObject == ConfigValidation.StringsValidator) stormConf.put(cleanedKey, Arrays.asList(conf.getStringArray(keyString))); else { logger.error("{} cannot be configured from XML. Consider configuring in navie storm configuration."); throw new UnsupportedOperationException(cleanedKey + " cannot be configured from XML"); } } }  static void main(String[] args);  }
@Test public void testConfigureStorm() throws Exception { URL resource = TestResourceUtils.getResource(this.getClass(), "storm-config.xml"); XMLConfiguration conf = new XMLConfiguration(resource); Config stormConf = new Config(); topology.configureStorm(conf, stormConf); assertThat(stormConf.get(TOPOLOGY_WORKERS), is(2)); assertThat(stormConf.get(TOPOLOGY_MESSAGE_TIMEOUT_SECS), is(300)); assertThat(stormConf.get(TOPOLOGY_MAX_SPOUT_PENDING), is(10)); assertThat(stormConf.get(TOPOLOGY_WORKER_CHILDOPTS), is(Arrays.asList("-Xmx4G"))); assertThat(stormConf.get(TOPOLOGY_DEBUG), is(true)); assertThat(stormConf.get(TOPOLOGY_STATS_SAMPLE_RATE), is(2.0d)); assertThat(stormConf.get(TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE), is(16L)); assertThat(stormConf.get(STORM_ZOOKEEPER_SERVERS), is(Arrays.asList("host0:2081", "host1:2081"))); assertThat(stormConf.get(STORM_LOCAL_DIR), is("/storm-local")); }
@Test public void testConfigureStormUnsupported(@Injectable Configuration conf) throws Exception { List<String> keys = Arrays.asList("topology..metrics..consumer..register"); new Expectations() {{ conf.getKeys(); result = keys.iterator(); }}; Config stormConf = new Config(); thrown.expect(UnsupportedOperationException.class); topology.configureStorm(conf, stormConf); }