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

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