MailBolt extends BaseRichBolt { public MailBolt() { } MailBolt(); @SuppressWarnings({ "rawtypes", "unchecked" }) @Override void prepare(Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple tuple); @Override void declareOutputFields(OutputFieldsDeclarer declarer); MailService getMailService();  }
@Test public void testMailBolt() { MailBolt bolt = new MailBolt(); final AtomicReference<Values> processedEventContainer = new AtomicReference<Values>(null); OutputCollector collector = MockTupleHelpers.mockCollector(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { Object newEvent = invocation.getArguments()[2]; processedEventContainer.set((Values) newEvent); System.out.println("Alert emitted:" + processedEventContainer.get()); return new ArrayList<>(); } }); bolt.prepare(null, null, collector); bolt.setMailService(service); Alert alert = new Alert(); when(service.sendMail(alert)).thenReturn(true); when(tuple.getValueByField(Constants.FIELD_ALERT)).thenReturn(alert); bolt.execute(tuple); verify(collector, times(1)).ack(tuple); verify(service, times(1)).sendMail(alert); }