CustomProcessorBolt extends AbstractProcessorBolt { @Override public void declareOutputFields (OutputFieldsDeclarer declarer) { declarer.declareStream(outputSchema.keySet().iterator().next(), new Fields(StreamlineEvent.STREAMLINE_EVENT)); }  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 testDeclareOutputFieldsWithoutOutputSchema () { customProcessorBolt.declareOutputFields(mockOutputDeclarer); }
@Test public void testDeclareOutputFields () { customProcessorBolt.outputSchema(outputStreamToSchema); customProcessorBolt.declareOutputFields(mockOutputDeclarer); new VerificationsInOrder(){{ mockOutputDeclarer.declareStream(outputStream, withAny(OUTPUT_FIELDS)); times = 1; }}; }