public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEvent event)
            throws Exception {
        mainThreadExecutor.assertRunningInMainThread();
        if (event instanceof AcknowledgeCheckpointEvent) {
            subtaskGatewayMap
                    .get(subtask)
                    .openGatewayAndUnmarkCheckpoint(
                            ((AcknowledgeCheckpointEvent) event).getCheckpointID());
            return;
        }
        coordinator.handleEventFromOperator(subtask, attemptNumber, event);
    }
@Test
    void acknowledgeCheckpointEventReleasesBlockedEvents() throws Exception {
        final EventReceivingTasks tasks = EventReceivingTasks.createForRunningTasks();
        final OperatorCoordinatorHolder holder =
                createCoordinatorHolder(tasks, TestingOperatorCoordinator::new);

        triggerAndCompleteCheckpoint(holder, 1111L);
        getCoordinator(holder).getSubtaskGateway(0).sendEvent(new TestOperatorEvent(1337));
        holder.handleEventFromOperator(0, 0, new AcknowledgeCheckpointEvent(1111L));

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