FilteringPluginSink implements PluginSink { @Override public void sendBatch(final Batch batch) { if (filter.matchesBatch(batch)) { sink.sendBatch(batch); } } FilteringPluginSink(final Filter filter); @Override void init(); @Override void sendMetric(final Metric metric); @Override void sendBatch(final Batch batch); @Override AsyncFuture<Void> start(); @Override AsyncFuture<Void> stop(); @Override boolean isReady(); PluginSink getSink();  }
@Test public void testSendBatchTrueFilter() { sink.filter = new TrueFilter(); doNothing().when(childSink).sendBatch(batch); ArgumentCaptor<Batch> captor = ArgumentCaptor.forClass(Batch.class); sink.sendBatch(batch); verify(childSink, times(1)).sendBatch(captor.capture()); Batch sentBatch = captor.getValue(); assertEquals("gew1", sentBatch.getCommonTags().get("pod")); assertEquals("stats", sentBatch.getCommonTags().get("what")); }