@Override
    public void open() throws Exception {
        if (useSplittableTimers()
                && areSplittableTimersConfigured()
                && getTimeServiceManager().isPresent()) {
            this.watermarkProcessor =
                    new MailboxWatermarkProcessor(
                            output, mailboxExecutor, getTimeServiceManager().get());
        }
    }
@Test
    void testStateDoesNotInterfere() throws Exception {
        try (KeyedOneInputStreamOperatorTestHarness<Integer, Tuple2<Integer, String>, String>
                testHarness = createTestHarness()) {
            testHarness.open();

            testHarness.processElement(new Tuple2<>(0, "SET_STATE:HELLO"), 0);
            testHarness.processElement(new Tuple2<>(1, "SET_STATE:CIAO"), 0);

            testHarness.processElement(new Tuple2<>(1, "EMIT_STATE"), 0);
            testHarness.processElement(new Tuple2<>(0, "EMIT_STATE"), 0);

            assertThat(extractResult(testHarness))
                    .contains("ON_ELEMENT:1:CIAO", "ON_ELEMENT:0:HELLO");
        }
    }