BoltWrapper extends AbstractStreamOperator<OUT> implements OneInputStreamOperator<IN, OUT> { @Override public void dispose() { this.bolt.cleanup(); } BoltWrapper(final IRichBolt bolt);  BoltWrapper(final IRichBolt bolt, final Fields inputSchema);  BoltWrapper(final IRichBolt bolt, final String[] rawOutputs);  BoltWrapper(final IRichBolt bolt, final Collection<String> rawOutputs);  BoltWrapper(final IRichBolt bolt, final Fields inputSchema, final String[] rawOutputs);  BoltWrapper(final IRichBolt bolt, final Fields inputSchema,
			final Collection<String> rawOutputs);  BoltWrapper(final IRichBolt bolt, final String name, final String inputStreamId,
			final String inputComponentId, final Fields inputSchema,
			final Collection<String> rawOutputs); void setStormTopology(StormTopology stormTopology); @Override void open(); @Override void dispose(); @Override void processElement(final StreamRecord<IN> element); @Override void processWatermark(Watermark mark); final static String DEFAULT_ID; final static String DEFUALT_BOLT_NAME; }
@SuppressWarnings("unchecked") @Test public void testClose() throws Exception { final IRichBolt bolt = mock(IRichBolt.class); final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer(); declarer.declare(new Fields("dummy")); PowerMockito.whenNew(SetupOutputFieldsDeclarer.class).withNoArguments().thenReturn(declarer); final BoltWrapper<Object, Object> wrapper = new BoltWrapper<Object, Object>(bolt); wrapper.setup(createMockStreamTask(), new StreamConfig(new Configuration()), mock(Output.class)); wrapper.close(); wrapper.dispose(); verify(bolt).cleanup(); }