ParserBolt extends ConfiguredParserBolt implements Serializable { @SuppressWarnings("unchecked") @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { super.prepare(stormConf, context, collector); messageGetStrategy = MessageGetters.DEFAULT_BYTES_FROM_POSITION.get(); this.collector = collector; this.parserRunner.init(this::getConfigurations, initializeStellar()); ackTuplesPolicy = new AckTuplesPolicy(collector, messageGetStrategy); for (Map.Entry<String, WriterHandler> entry: sensorToWriterMap.entrySet()) { String sensor = entry.getKey(); SensorParserConfig config = getSensorParserConfig(sensor); if (config != null) { config.init(); topicToSensorMap.put(config.getSensorTopic(), sensor); } else { throw new IllegalStateException( "Unable to retrieve a parser config for " + sensor); } WriterHandler writer = sensorToWriterMap.get(sensor); if (maxBatchTimeout == 0) { WriterConfiguration writerConfig = getConfigurationStrategy() .createWriterConfig(writer.getBulkMessageWriter(), getConfigurations()); BatchTimeoutHelper timeoutHelper = new BatchTimeoutHelper( writerConfig::getAllConfiguredTimeouts, batchTimeoutDivisor); maxBatchTimeout = timeoutHelper.getMaxBatchTimeout(); } writer.init(stormConf, context, collector, getConfigurations(), ackTuplesPolicy, maxBatchTimeout); } } 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 shouldThrowExceptionOnMissingConfig() { Map stormConf = mock(Map.class); ParserBolt parserBolt = new ParserBolt("zookeeperUrl", parserRunner, new HashMap<String, WriterHandler>() {{ put("yaf", writerHandler); }}); parserBolt.setCuratorFramework(client); parserBolt.setZKCache(cache); IllegalStateException e = assertThrows(IllegalStateException.class, () -> parserBolt.prepare(stormConf, topologyContext, outputCollector)); assertEquals("Unable to retrieve a parser config for yaf", e.getMessage()); }