ElasticSearchJsonBolt extends BaseBasicBolt implements Configurable { Map<String, String> getFieldTypeMapping(Map<String, Map<String, String>> fieldTypeMappings, String cleanedKey) { boolean hasFieldMapping = fieldTypeMappings.containsKey(cleanedKey); Map<String, String> fieldTypeMapping = null; if (hasFieldMapping) { fieldTypeMapping = fieldTypeMappings.get(cleanedKey); } else { fieldTypeMapping = Collections.emptyMap(); } return fieldTypeMapping; }  @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 testGetFieldTypeMappingEmpty() throws Exception { URL resource = getResource("configureFieldTypeMapping.xml"); XMLConfiguration conf = new XMLConfiguration(resource); Map<String, Map<String, String>> fieldTypeMappings = extractMapList(conf.subset("fieldTypeMapping"), "entry", FIELD_NAME, FIELD_TYPE, DATE_FORMAT); assertThat(fieldTypeMappings.size(), is(2)); Map<String, String> mapping = bolt.getFieldTypeMapping(fieldTypeMappings, "field_that_does_not_exist"); assertThat(mapping.isEmpty(), is(true)); }
@Test public void testGetFieldTypeMapping() throws Exception { URL resource = getResource("configureFieldTypeMapping.xml"); XMLConfiguration conf = new XMLConfiguration(resource); Map<String, Map<String, String>> fieldTypeMappings = extractMapList(conf.subset("fieldTypeMapping"), "entry", FIELD_NAME, FIELD_TYPE, DATE_FORMAT); assertThat(fieldTypeMappings.size(), is(2)); Map<String, String> mapping = bolt.getFieldTypeMapping(fieldTypeMappings, "geo_coordinates"); assertThat(mapping.get(ElasticSearchJsonBolt.FIELD_TYPE), is("array")); }