HBaseBolt extends BaseRichBolt { @Override public void execute(Tuple tuple) { LOG.trace("Received a tuple."); try { if (batchHelper.shouldHandle(tuple)) { save(tuple); } if (batchHelper.shouldFlush()) { flush(); } } catch (Exception e) { batchHelper.fail(e); hbaseClient.clearMutations(); } } HBaseBolt(String tableName, HBaseMapper mapper); HBaseBolt writeToWAL(boolean writeToWAL); HBaseBolt withTableProvider(String tableProvider); HBaseBolt withTableProviderInstance(TableProvider tableProvider); HBaseBolt withBatchSize(int batchSize); HBaseBolt withFlushIntervalSecs(int flushIntervalSecs); void setClient(HBaseClient hbaseClient); @Override Map<String, Object> getComponentConfiguration(); @Override void prepare(Map map, TopologyContext topologyContext, OutputCollector collector); @Override void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer); @Override void execute(Tuple tuple);  }
@Test public void testBatchReady() { HBaseBolt bolt = createBolt(2, new WidgetMapper()); bolt.execute(tuple1); bolt.execute(tuple2); verify(client, times(2)).addMutation(any(), any(), any()); verify(client, times(1)).mutate(); }
@Test public void testBatchNotReady() { HBaseBolt bolt = createBolt(2, new WidgetMapper()); bolt.execute(tuple1); verify(client, times(1)).addMutation(any(), any(), any()); verify(client, times(0)).mutate(); }
@Test public void testTimeFlush() { HBaseBolt bolt = createBolt(2, new WidgetMapper()); bolt.execute(tuple1); verify(client, times(1)).addMutation(any(), any(), any()); verify(client, times(0)).mutate(); bolt.execute(mockTickTuple()); verify(client, times(1)).mutate(); }
@Test public void testWriteWithTTL() { final Long expectedTTL = 2000L; WidgetMapper mapperWithTTL = new WidgetMapper(expectedTTL); HBaseBolt bolt = createBolt(2, mapperWithTTL); bolt.execute(tuple1); bolt.execute(tuple2); ArgumentCaptor<Long> ttlCaptor = ArgumentCaptor.forClass(Long.class); verify(client, times(2)).addMutation(any(), any(), any(), ttlCaptor.capture()); assertEquals(expectedTTL, ttlCaptor.getValue()); }