public void emitWatermarkInsideMailbox(Watermark mark) throws Exception {
        maxInputWatermark =
                new Watermark(Math.max(maxInputWatermark.getTimestamp(), mark.getTimestamp()));
        emitWatermarkInsideMailbox();
    }
@Test
    void testEmitWatermarkInsideMailbox() throws Exception {
        int priority = 42;
        final List<StreamElement> emittedElements = new ArrayList<>();
        final TaskMailboxImpl mailbox = new TaskMailboxImpl();
        final InternalTimeServiceManager<?> timerService = new NoOpInternalTimeServiceManager();

        final MailboxWatermarkProcessor<StreamRecord<String>> watermarkProcessor =
                new MailboxWatermarkProcessor<>(
                        new CollectorOutput<>(emittedElements),
                        new MailboxExecutorImpl(
                                mailbox, priority, StreamTaskActionExecutor.IMMEDIATE),
                        timerService);
        final List<Watermark> expectedOutput = new ArrayList<>();
        watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(1));
        watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(2));
        watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(3));
        expectedOutput.add(new Watermark(1));
        expectedOutput.add(new Watermark(2));
        expectedOutput.add(new Watermark(3));

        assertThat(emittedElements).containsExactlyElementsOf(expectedOutput);

        mailbox.put(new Mail(() -> {}, TaskMailbox.MIN_PRIORITY, "checkpoint mail"));

        watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(4));
        watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(5));

        assertThat(emittedElements).containsExactlyElementsOf(expectedOutput);

        // FLINK-35528: do not allow yielding to continuation mails
        assertThat(mailbox.tryTake(priority)).isEqualTo(Optional.empty());
        assertThat(emittedElements).containsExactlyElementsOf(expectedOutput);

        while (mailbox.hasMail()) {
            mailbox.take(TaskMailbox.MIN_PRIORITY).run();
        }
        // Watermark(4) is processed together with Watermark(5)
        expectedOutput.add(new Watermark(5));

        assertThat(emittedElements).containsExactlyElementsOf(expectedOutput);
    }