BulkMessageWriterBolt extends ConfiguredBolt<CONFIG_T> { @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declareStream(Constants.ERROR_STREAM, new Fields("message")); } BulkMessageWriterBolt(String zookeeperUrl, String configurationStrategy); BulkMessageWriterBolt<CONFIG_T> withBulkMessageWriter(BulkMessageWriter<JSONObject> bulkMessageWriter); BulkMessageWriterBolt<CONFIG_T> withMessageWriter(MessageWriter<JSONObject> messageWriter); BulkMessageWriterBolt<CONFIG_T> withMessageGetter(String messageGetStrategyType); BulkMessageWriterBolt<CONFIG_T> withMessageGetterField(String messageGetField); BulkMessageWriterBolt<CONFIG_T> withBatchTimeoutDivisor(int batchTimeoutDivisor); int getMaxBatchTimeout(); BulkWriterComponent<JSONObject> getWriterComponent(); void setWriterComponent(BulkWriterComponent<JSONObject> component); @Override Map<String, Object> getComponentConfiguration(); @Override void prepare(Map stormConf, TopologyContext context, OutputCollector collector); void prepare(Map stormConf, TopologyContext context, OutputCollector collector, Clock clock); @SuppressWarnings("unchecked") @Override void execute(Tuple tuple); @Override void declareOutputFields(OutputFieldsDeclarer declarer);  }
@Test public void testDeclareOutputFields() { BulkMessageWriterBolt<IndexingConfigurations> bulkMessageWriterBolt = new BulkMessageWriterBolt<IndexingConfigurations>( "zookeeperUrl", "INDEXING"); bulkMessageWriterBolt.declareOutputFields(declarer); verify(declarer, times(1)).declareStream(eq("error") , argThat(new FieldsMatcher("message"))); }