@Test public void testCleanup() throws Exception { BoltOutputBatcher bolt = new BoltOutputBatcher(this.boltMock, this.noBatching); BatchOutputCollector collectorMock = mock(BatchOutputCollector.class); PowerMockito.whenNew(BatchOutputCollector.class).withAnyArguments().thenReturn(collectorMock); bolt.prepare(null, null, null); bolt.cleanup(); verify(this.boltMock).cleanup(); verify(collectorMock).flush(); }
@Override public void cleanup() { this.wrappedBolt.cleanup(); this.batchCollector.flush(); }
BoltOutputBatcher implements IRichBolt { @Override public void cleanup() { this.wrappedBolt.cleanup(); this.batchCollector.flush(); } }
BoltOutputBatcher implements IRichBolt { @Override public void cleanup() { this.wrappedBolt.cleanup(); this.batchCollector.flush(); } BoltOutputBatcher(IRichBolt bolt, int batchSize);  BoltOutputBatcher(IRichBolt bolt, Map<String, Integer> batchSizes); }
BoltOutputBatcher implements IRichBolt { @Override public void cleanup() { this.wrappedBolt.cleanup(); this.batchCollector.flush(); } BoltOutputBatcher(IRichBolt bolt, int batchSize);  BoltOutputBatcher(IRichBolt bolt, Map<String, Integer> batchSizes); @Override void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void cleanup(); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override Map<String, Object> getComponentConfiguration(); static void registerKryoClasses(Config stormConfig); }
BoltOutputBatcher implements IRichBolt { @Override public void cleanup() { this.wrappedBolt.cleanup(); this.batchCollector.flush(); } BoltOutputBatcher(IRichBolt bolt, int batchSize);  BoltOutputBatcher(IRichBolt bolt, Map<String, Integer> batchSizes); @Override void prepare(@SuppressWarnings("rawtypes") Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void cleanup(); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override Map<String, Object> getComponentConfiguration(); static void registerKryoClasses(Config stormConfig);  }