EsLookupBolt extends AbstractEsBolt { @Override public void execute(Tuple tuple) { try { Collection<Values> values = lookupValuesInEs(tuple); tryEmitAndAck(values, tuple); } catch (Exception e) { collector.reportError(e); collector.fail(tuple); } } EsLookupBolt(EsConfig esConfig, ElasticsearchGetRequest getRequest, EsLookupResultOutput output); @Override void execute(Tuple tuple); @Override void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer);  }
@Test public void failsTupleWhenClientThrows() throws Exception { when(client.get(request)).thenThrow(ElasticsearchException.class); bolt.execute(tuple); verify(outputCollector).fail(tuple); }
@Test public void reportsExceptionWhenClientThrows() throws Exception { ElasticsearchException elasticsearchException = new ElasticsearchException("dummy"); when(client.get(request)).thenThrow(elasticsearchException); bolt.execute(tuple); verify(outputCollector).reportError(elasticsearchException); }