ParserBolt extends ConfiguredParserBolt implements Serializable { @Override public Map<String, Object> getComponentConfiguration() { Function<WriterConfiguration, WriterConfiguration> configurationXform; WriterHandler writer = sensorToWriterMap.entrySet().iterator().next().getValue(); if (writer.isWriterToBulkWriter()) { configurationXform = WriterToBulkWriter.TRANSFORMATION; } else { configurationXform = x -> x; } WriterConfiguration writerconf = configurationXform .apply(getConfigurationStrategy() .createWriterConfig(writer.getBulkMessageWriter(), getConfigurations())); BatchTimeoutHelper timeoutHelper = new BatchTimeoutHelper(writerconf::getAllConfiguredTimeouts, batchTimeoutDivisor); this.requestedTickFreqSecs = timeoutHelper.getRecommendedTickInterval(); this.maxBatchTimeout = timeoutHelper.getMaxBatchTimeout(); Map<String, Object> conf = super.getComponentConfiguration(); if (conf == null) { conf = new HashMap<>(); } conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, requestedTickFreqSecs); LOG.info("Requesting " + Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS + " set to " + Integer.toString(requestedTickFreqSecs)); return conf; } ParserBolt( String zookeeperUrl
                   , ParserRunner parserRunner
                     , Map<String, WriterHandler> sensorToWriterMap
  ); ParserBolt withBatchTimeoutDivisor(int batchTimeoutDivisor); int getBatchTimeoutDivisor(); void setMessageGetStrategy(MessageGetStrategy messageGetStrategy); void setOutputCollector(OutputCollector collector); void setAckTuplesPolicy(AckTuplesPolicy ackTuplesPolicy); @Override Map<String, Object> getComponentConfiguration(); @SuppressWarnings("unchecked") @Override void prepare(Map stormConf, TopologyContext context, OutputCollector collector); @SuppressWarnings("unchecked") @Override void execute(Tuple tuple); @Override void declareOutputFields(OutputFieldsDeclarer declarer);  }
@Test public void shouldGetComponentConfiguration() { ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserRunner, new HashMap<String, WriterHandler>() {{ put("yaf", writerHandler); }}) { @Override public ParserConfigurations getConfigurations() { ParserConfigurations configurations = new ParserConfigurations(); SensorParserConfig sensorParserConfig = new SensorParserConfig(); sensorParserConfig.setParserConfig(new HashMap<String, Object>() {{ put(IndexingConfigurations.BATCH_SIZE_CONF, 10); }}); configurations.updateSensorParserConfig("yaf", sensorParserConfig); return configurations; } }; Map<String, Object> componentConfiguration = parserBolt.getComponentConfiguration(); assertEquals(1, componentConfiguration.size()); assertEquals( 14, componentConfiguration.get(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS)); }