StreamFilter extends AbstractUdfStreamOperator<IN, FilterFunction<IN>> implements OneInputStreamOperator<IN, IN> { @Override public void processElement(StreamRecord<IN> element) throws Exception { if (userFunction.filter(element.getValue())) { output.collect(element); } } StreamFilter(FilterFunction<IN> filterFunction); @Override void processElement(StreamRecord<IN> element); @Override void processWatermark(Watermark mark);  }
@Test public void testOpenClose() throws Exception { StreamFilter<String> operator = new StreamFilter<String>(new TestOpenCloseFilterFunction()); OneInputStreamOperatorTestHarness<String, String> testHarness = new OneInputStreamOperatorTestHarness<String, String>(operator); long initialTime = 0L; testHarness.open(); testHarness.processElement(new StreamRecord<String>("fooHello", initialTime)); testHarness.processElement(new StreamRecord<String>("bar", initialTime)); testHarness.close(); Assert.assertTrue("RichFunction methods where not called.", TestOpenCloseFilterFunction.closeCalled); Assert.assertTrue("Output contains no elements.", testHarness.getOutput().size() > 0); }