ElasticSearchJsonBolt extends BaseBasicBolt implements Configurable { void configureFieldTypeMapping(Configuration conf) { fieldTypeMappings = extractMapList(conf, "entry", FIELD_NAME, FIELD_TYPE, DATE_FORMAT); }  @Override void configure(Configuration conf); @Override void prepare(Map stormConf, TopologyContext context); @Override final void declareOutputFields(OutputFieldsDeclarer declarer); @Override final void execute(Tuple input, BasicOutputCollector collector); static final String ES_JSON; }
@Test public void testConfigureFieldTypeMapping() throws Exception { URL resource = getResource("configureFieldTypeMapping.xml"); XMLConfiguration conf = new XMLConfiguration(resource); bolt.configureFieldTypeMapping(conf.subset("fieldTypeMapping")); assertThat(bolt.fieldTypeMappings.size(), is(2)); Map<String, String> geo_coordinates = bolt.fieldTypeMappings.get("geo_coordinates"); assertThat(geo_coordinates.get(ElasticSearchJsonBolt.FIELD_NAME), is("geo_coordinates")); assertThat(geo_coordinates.get(ElasticSearchJsonBolt.FIELD_TYPE), is("array")); }