StreamMap extends AbstractUdfStreamOperator<OUT, MapFunction<IN, OUT>> implements OneInputStreamOperator<IN, OUT> { @Override public void processElement(StreamRecord<IN> element) throws Exception { output.collect(element.replace(userFunction.map(element.getValue()))); } StreamMap(MapFunction<IN, OUT> mapper); @Override void processElement(StreamRecord<IN> element); @Override void processWatermark(Watermark mark);  }
@Test public void testOpenClose() throws Exception { StreamMap<String, String> operator = new StreamMap<String, String>(new TestOpenCloseMapFunction()); OneInputStreamOperatorTestHarness<String, String> testHarness = new OneInputStreamOperatorTestHarness<String, String>(operator); long initialTime = 0L; testHarness.open(); testHarness.processElement(new StreamRecord<String>("Hello", initialTime)); testHarness.close(); Assert.assertTrue("RichFunction methods where not called.", TestOpenCloseMapFunction.closeCalled); Assert.assertTrue("Output contains no elements.", testHarness.getOutput().size() > 0); }