ElasticSearchJsonBolt extends BaseBasicBolt implements Configurable { void configureTimeSeriesIndex(Configuration timeSeriesIndexNameConfig) { if (!timeSeriesIndexNameConfig.isEmpty()) { timeSeriesIndexFieldName = timeSeriesIndexNameConfig.getString(FIELD_NAME); timeSeriesIndexInputDateFormat = timeSeriesIndexNameConfig.getString(INPUT_DATE_FORMAT); timeSeriesIndexOutputDateFormat = timeSeriesIndexNameConfig.getString(POSTFIX_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 testConfigureTimeSeriesIndex() throws Exception { URL resource = getResource("configureTimeSeriesIndex.xml"); XMLConfiguration conf = new XMLConfiguration(resource); bolt.configureTimeSeriesIndex(conf.subset("timeSeriesIndex")); assertThat(bolt.timeSeriesIndexFieldName, is("postedTime")); assertThat(bolt.timeSeriesIndexInputDateFormat, is("yyyy-MM-dd'T'HH:mm:ss.SSSX")); assertThat(bolt.timeSeriesIndexOutputDateFormat, is("yyyy.MM.dd.HH")); }
@Test public void testConfigureTimeSeriesIndexDefault() throws Exception { URL resource = getResource("configureDefault.xml"); XMLConfiguration conf = new XMLConfiguration(resource); bolt.configureTimeSeriesIndex(conf.subset("timeSeriesIndex")); assertNull(bolt.timeSeriesIndexFieldName); assertNull(bolt.timeSeriesIndexInputDateFormat); assertNull(bolt.timeSeriesIndexOutputDateFormat); }