@Override
    public void resetToCheckpoint(long checkpointId, @Nullable byte[] checkpointData)
            throws Exception {
        // the first time this method is called is early during execution graph construction,
        // before the main thread executor is set. hence this conditional check.
        if (mainThreadExecutor != null) {
            mainThreadExecutor.assertRunningInMainThread();
        }

        subtaskGatewayMap.values().forEach(SubtaskGatewayImpl::openGatewayAndUnmarkAllCheckpoint);
        context.resetFailed();

        // when initial savepoints are restored, this call comes before the mainThreadExecutor
        // is available, which is needed to set up these gateways. So during the initial restore,
        // we ignore this, and instead the gateways are set up in the "lazyInitialize" method, which
        // is called when the scheduler is properly set up.
        // this is a bit clumsy, but it is caused by the non-straightforward initialization of the
        // ExecutionGraph and Scheduler.
        if (mainThreadExecutor != null) {
            setupAllSubtaskGateways();
        }

        coordinator.resetToCheckpoint(checkpointId, checkpointData);
    }
@Test
    void restoreOpensGatewayEvents() throws Exception {
        final EventReceivingTasks tasks = EventReceivingTasks.createForRunningTasks();
        final OperatorCoordinatorHolder holder =
                createCoordinatorHolder(tasks, TestingOperatorCoordinator::new);

        triggerAndCompleteCheckpoint(holder, 1000L);
        holder.resetToCheckpoint(1L, new byte[0]);
        getCoordinator(holder).getSubtaskGateway(1).sendEvent(new TestOperatorEvent(999));

        assertThat(tasks.getSentEventsForSubtask(1)).containsExactly(new TestOperatorEvent(999));
    }