@Override
    public void processElement(StreamRecord<IN> element) throws Exception {
        output.collect(element.replace(userFunction.map(element.getValue())));
    }
@Test
    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();

        assertThat(TestOpenCloseMapFunction.closeCalled)
                .as("RichFunction methods where not called.")
                .isTrue();
        assertThat(testHarness.getOutput()).as("Output contains no elements.").isNotEmpty();
    }