@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);
}