public static DelayableTimer createRegisterTimerCallback(
            MailboxExecutor mailboxExecutor, TimerService timerService) {
        return (callable, delay) -> {
            ScheduledFuture<?> scheduledFuture =
                    timerService.registerTimer(
                            timerService.getCurrentProcessingTime() + delay.toMillis(),
                            timestamp ->
                                    mailboxExecutor.execute(
                                            () -> callable.call(),
                                            "Execute checkpoint barrier handler delayed action"));
            return () -> scheduledFuture.cancel(false);
        };
    }
@Test
    void testDelayableTimerNotHiddenException() throws Exception {
        TaskMailbox mailbox = new TaskMailboxImpl();
        MailboxProcessor mailboxProcessor =
                new MailboxProcessor(controller -> {}, mailbox, StreamTaskActionExecutor.IMMEDIATE);
        MailboxExecutor mailboxExecutor =
                new MailboxExecutorImpl(
                        mailbox, 0, StreamTaskActionExecutor.IMMEDIATE, mailboxProcessor);

        TestProcessingTimeService timerService = new TestProcessingTimeService();
        timerService.setCurrentTime(System.currentTimeMillis());

        BarrierAlignmentUtil.DelayableTimer delayableTimer =
                BarrierAlignmentUtil.createRegisterTimerCallback(mailboxExecutor, timerService);

        Duration delay = Duration.ofMinutes(10);

        delayableTimer.registerTask(
                () -> {
                    // simulate Exception in checkpoint sync phase
                    throw new ExpectedTestException();
                },
                delay);

        timerService.advance(delay.toMillis());

        assertThatThrownBy(mailboxProcessor::runMailboxStep)
                .as("BarrierAlignmentUtil.DelayableTimer should not hide exceptions")
                .isInstanceOf(ExpectedTestException.class);
    }