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

            testHarness.setProcessingTime(0L);

            testHarness.processElement(new Tuple2<>(1, "SET_PROC_TIME_TIMER:20"), 0);

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

            testHarness.processElement(new Tuple2<>(0, "SET_PROC_TIME_TIMER:10"), 0);

            testHarness.setProcessingTime(10L);

            assertThat(extractResult(testHarness)).contains("ON_PROC_TIME:HELLO");

            testHarness.setProcessingTime(20L);

            assertThat(extractResult(testHarness)).contains("ON_PROC_TIME:CIAO");
        }
    }