@Override
    public void checkpointCoordinator(long checkpointId, CompletableFuture<byte[]> result) {
        // unfortunately, this method does not run in the scheduler executor, but in the
        // checkpoint coordinator time thread.
        // we can remove the delegation once the checkpoint coordinator runs fully in the
        // scheduler's main thread executor
        mainThreadExecutor.execute(() -> checkpointCoordinatorInternal(checkpointId, result));
    }
@Test
    void testCheckpointFailsIfSendingEventFailedBeforeTrigger() throws Exception {
        final ReorderableManualExecutorService executor = new ReorderableManualExecutorService();
        final ComponentMainThreadExecutor mainThreadExecutor =
                new ComponentMainThreadExecutorServiceAdapter(
                        (ScheduledExecutorService) executor, Thread.currentThread());

        CompletableFuture<Acknowledge> eventSendingResult = new CompletableFuture<>();
        final EventReceivingTasks tasks =
                EventReceivingTasks.createForRunningTasksWithRpcResult(eventSendingResult);

        final OperatorCoordinatorHolder holder =
                createCoordinatorHolder(tasks, TestingOperatorCoordinator::new, mainThreadExecutor);

        // Send one event without finishing it.
        getCoordinator(holder).getSubtaskGateway(0).sendEvent(new TestOperatorEvent(0));
        executor.triggerAll();

        // Finish the event sending. This will insert one runnable that handles
        // failed events to the executor. And we delay this runnable to
        // simulates checkpoints triggered before the failure get processed.
        executor.setDelayNewRunnables(true);
        eventSendingResult.completeExceptionally(new RuntimeException("Artificial"));
        executor.setDelayNewRunnables(false);

        // Trigger one checkpoint, the checkpoint should not be confirmed
        // before the failure get triggered.
        CompletableFuture<byte[]> checkpointResult = new CompletableFuture<>();
        holder.checkpointCoordinator(1, checkpointResult);
        executor.triggerAll();
        getCoordinator(holder).getLastTriggeredCheckpoint().complete(new byte[0]);
        executor.triggerAll();
        assertThat(checkpointResult).isNotDone();

        // Then the failure finally get processed by fail the corresponding tasks.
        executor.executeAllDelayedRunnables();
        executor.triggerAll();

        // The checkpoint would be finally confirmed.
        assertThat(checkpointResult).isCompletedExceptionally();
    }