@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 testCheckpointFailsIfSendingEventFailedAfterTrigger() throws Exception {
        CompletableFuture<Acknowledge> eventSendingResult = new CompletableFuture<>();
        final EventReceivingTasks tasks =
                EventReceivingTasks.createForRunningTasksWithRpcResult(eventSendingResult);
        final OperatorCoordinatorHolder holder =
                createCoordinatorHolder(tasks, TestingOperatorCoordinator::new);

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

        // Trigger one checkpoint.
        CompletableFuture<byte[]> checkpointResult = new CompletableFuture<>();
        holder.checkpointCoordinator(1, checkpointResult);
        getCoordinator(holder).getLastTriggeredCheckpoint().complete(new byte[0]);

        // Fail the event sending.
        eventSendingResult.completeExceptionally(new RuntimeException("Artificial"));

        assertThat(checkpointResult).isCompletedExceptionally();
    }