Sampler extends BaseKeyOperator<K> { public void setSamplingPercentage(double samplingPercentage) { this.samplingPercentage = samplingPercentage; }  double getSamplingPercentage(); void setSamplingPercentage(double samplingPercentage); final transient DefaultInputPort<K> data; final transient DefaultOutputPort<K> sample; }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testNodeProcessing() throws Exception { Sampler<String> oper = new Sampler<String>(); CountTestSink sink = new CountTestSink<String>(); oper.sample.setSink(sink); oper.setSamplingPercentage(.1); String tuple = "a"; int numTuples = 10000; oper.beginWindow(0); for (int i = 0; i < numTuples; i++) { oper.data.process(tuple); } oper.endWindow(); int lowerlimit = 5; int upperlimit = 15; int actual = (100 * sink.count) / numTuples; Assert.assertEquals("number emitted tuples", true, lowerlimit < actual); Assert.assertEquals("number emitted tuples", true, upperlimit > actual); }