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 takeCheckpointAfterSuccessfulCheckpoint() throws Exception {
        final EventReceivingTasks tasks = EventReceivingTasks.createForRunningTasks();
        final OperatorCoordinatorHolder holder =
                createCoordinatorHolder(tasks, TestingOperatorCoordinator::new);

        getCoordinator(holder).getSubtaskGateway(0).sendEvent(new TestOperatorEvent(0));

        triggerAndCompleteCheckpoint(holder, 22L);
        getCoordinator(holder).getSubtaskGateway(0).sendEvent(new TestOperatorEvent(1));
        holder.handleEventFromOperator(0, 0, new AcknowledgeCheckpointEvent(22L));
        holder.handleEventFromOperator(1, 0, new AcknowledgeCheckpointEvent(22L));
        holder.handleEventFromOperator(2, 0, new AcknowledgeCheckpointEvent(22L));

        getCoordinator(holder).getSubtaskGateway(0).sendEvent(new TestOperatorEvent(2));

        triggerAndCompleteCheckpoint(holder, 23L);
        getCoordinator(holder).getSubtaskGateway(0).sendEvent(new TestOperatorEvent(3));
        holder.handleEventFromOperator(0, 0, new AcknowledgeCheckpointEvent(23L));
        holder.handleEventFromOperator(1, 0, new AcknowledgeCheckpointEvent(23L));
        holder.handleEventFromOperator(2, 0, new AcknowledgeCheckpointEvent(23L));

        assertThat(tasks.getSentEventsForSubtask(0))
                .containsExactly(
                        new TestOperatorEvent(0),
                        new TestOperatorEvent(1),
                        new TestOperatorEvent(2),
                        new TestOperatorEvent(3));
    }