@Override
    public void abortCurrentTriggering() {
        // 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(
                () ->
                        subtaskGatewayMap
                                .values()
                                .forEach(
                                        SubtaskGatewayImpl
                                                ::openGatewayAndUnmarkLastCheckpointIfAny));
    }
@Test
    void abortedCheckpointReleasesBlockedEvents() throws Exception {
        final EventReceivingTasks tasks = EventReceivingTasks.createForRunningTasks();
        final OperatorCoordinatorHolder holder =
                createCoordinatorHolder(tasks, TestingOperatorCoordinator::new);

        triggerAndCompleteCheckpoint(holder, 123L);
        getCoordinator(holder).getSubtaskGateway(0).sendEvent(new TestOperatorEvent(1337));
        holder.abortCurrentTriggering();

        assertThat(tasks.getSentEventsForSubtask(0)).containsExactly(new TestOperatorEvent(1337));
    }