CustomProcessorBolt extends AbstractProcessorBolt { @Override public void prepare (Map stormConf, TopologyContext context, OutputCollector collector) { super.prepare(stormConf, context, collector); customProcessorRuntime = getCustomProcessorRuntime(); customProcessorRuntime.initialize(config); }  CustomProcessorBolt customProcessorImpl(String customProcessorImpl); CustomProcessorBolt outputSchema(String outputSchemaJson); CustomProcessorBolt outputSchema(Map<String, Schema> outputSchema); CustomProcessorBolt inputSchemaMap(Map<String, Map<String, String>> inputSchemaMap); CustomProcessorBolt inputSchemaMap(String inputSchemaMapJson); CustomProcessorBolt config(String configJson); CustomProcessorBolt config(Map<String, Object> config); @Override void prepare(Map stormConf, TopologyContext context, OutputCollector collector); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override void cleanup();  }
@Test(expected = RuntimeException.class) public void testNoCustomImplPrepare () throws Exception { customProcessorBolt.prepare(new HashMap(), null, null); }
@Test public void testPrepare () throws ClassNotFoundException, MalformedURLException, InstantiationException, IllegalAccessException { customProcessorBolt.customProcessorImpl(ConsoleCustomProcessor.class.getCanonicalName()); final Map<String, Object> config = new HashMap<>(); customProcessorBolt.config(config); new Expectations() {{ customProcessorRuntime.initialize(withEqual(config)); minTimes=1; maxTimes=1; }}; Map<Object, Object> conf = new HashMap<>(); customProcessorBolt.prepare(conf, null, null); new VerificationsInOrder(){{ customProcessorRuntime.initialize(config); times = 1; }}; }