MetricStreamPersist extends BaseRichBolt { public MetricStreamPersist(MetricDescriptor metricDescriptor, Config config) { this.config = config; this.mapper = new StructuredMetricMapper(metricDescriptor); this.batchSize = config.hasPath("service.batchSize") ? config.getInt("service.batchSize") : 1; } MetricStreamPersist(MetricDescriptor metricDescriptor, Config config);  MetricStreamPersist(MetricMapper mapper, Config config); @Override void prepare(Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override void cleanup(); static final String METRIC_NAME_FIELD; static final String METRIC_EVENT_FIELD; }
@Test public void testMetricStreamPersist() throws Exception { List<String> result = new ArrayList<>(); OutputCollector collector = new OutputCollector(new IOutputCollector() { @Override public List<Integer> emit(String streamId, Collection<Tuple> anchors, List<Object> tuple) { result.add(String.valueOf(tuple.get(0))); return null; } @Override public void emitDirect(int taskId, String streamId, Collection<Tuple> anchors, List<Object> tuple) { } @Override public void ack(Tuple input) { } @Override public void fail(Tuple input) { } @Override public void reportError(Throwable error) { } }); MetricDescriptor metricDefinition = MetricDescriptor .metricGroupByField("group") .siteAs("siteId") .namedByField("metric") .eventTimeByField("timestamp") .dimensionFields("host", "component", "site") .granularity(Calendar.MINUTE) .valueField("value"); Config config = mock(Config.class); when(config.hasPath("service.batchSize")).thenReturn(false); GenericServiceAPIResponseEntity<String> response = mock(GenericServiceAPIResponseEntity.class); when(response.isSuccess()).thenReturn(true); EagleServiceClientImpl client = mock(EagleServiceClientImpl.class); PowerMockito.whenNew(EagleServiceClientImpl.class).withArguments(config).thenReturn(client); when(client.create(anyObject())).thenReturn(response); MetricStreamPersist metricStreamPersist = new MetricStreamPersist(metricDefinition, config); metricStreamPersist.prepare(null, null, collector); Map event = new HashMap(); event.put("timestamp", 1482106479564L); event.put("metric", "hadoop.memory.heapmemoryusage.used"); event.put("component", "hbasemaster"); event.put("site", "sandbox"); event.put("value", 14460904.0); event.put("host", "xxx-xxx.int.xxx.com"); Tuple tuple = Testing.testTuple(new Values("metric", event)); metricStreamPersist.execute(tuple); Assert.assertTrue(result.size() == 1); Assert.assertEquals("hadoop.memory.heapmemoryusage.used", result.get(0)); }