@Override
    public void processElement(StreamRecord<IN> element) throws Exception {
        IN value = element.getValue();
        IN currentValue = values.value();

        if (currentValue == null) {
            // register a timer for emitting the result at the end when this is the
            // first input for this key
            timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, Long.MAX_VALUE);
        } else {
            // otherwise, reduce things
            value = userFunction.reduce(currentValue, value);
        }
        values.update(value);
    }
@Test
    void resultsOnMaxWatermark() throws Exception {
        KeyedOneInputStreamOperatorTestHarness<String, String, String> testHarness =
                createTestHarness();

        testHarness.processElement(new StreamRecord<>("hello"));
        testHarness.processElement(new StreamRecord<>("hello"));
        testHarness.processElement(new StreamRecord<>("ciao"));
        testHarness.processElement(new StreamRecord<>("ciao"));
        testHarness.processElement(new StreamRecord<>("ciao"));

        testHarness.processWatermark(Long.MAX_VALUE);

        ArrayDeque<Object> expectedOutput = new ArrayDeque<>();
        expectedOutput.add(new StreamRecord<>("hellohello", Long.MAX_VALUE));
        expectedOutput.add(new StreamRecord<>("ciaociaociao", Long.MAX_VALUE));
        expectedOutput.add(new Watermark(Long.MAX_VALUE));

        assertThat(testHarness.getOutput()).contains(expectedOutput.toArray());
    }