ElasticSearchJsonBolt extends BaseBasicBolt implements Configurable { @Override final public void execute(Tuple input, BasicOutputCollector collector) { LogRecord record = (LogRecord) input.getValueByField(AbstractLogRecordBolt.RECORD); try { indexRecord(record); } catch (Exception e) { logger.error("Error indexing record", e); throw new FailedException("Error indexing record", e); } collector.emit(new Values(record)); collector.emit(ES_JSON, new Values(record.getValue(esJsonField))); }  @Override void configure(Configuration conf); @Override void prepare(Map stormConf, TopologyContext context); @Override final void declareOutputFields(OutputFieldsDeclarer declarer); @Override final void execute(Tuple input, BasicOutputCollector collector); static final String ES_JSON; }
@Test public void testExecute( @Injectable Tuple input, @Injectable BasicOutputCollector collector, @Injectable LogRecord logRecord) throws Exception { bolt.esJsonField = "field"; new Expectations(bolt) {{ input.getValueByField(AbstractLogRecordBolt.RECORD); result = logRecord; bolt.indexRecord(logRecord); logRecord.getValue(bolt.esJsonField); result = "value"; collector.emit(new Values(logRecord)); collector.emit(ElasticSearchJsonBolt.ES_JSON, new Values("value")); }}; bolt.execute(input, collector); }