@Override
    public void deliverOperatorEventToCoordinator(
            ExecutionAttemptID taskExecution, OperatorID operator, OperatorEvent evt)
            throws FlinkException {
        final StateWithExecutionGraph stateWithExecutionGraph =
                state.as(StateWithExecutionGraph.class)
                        .orElseThrow(
                                () ->
                                        new TaskNotRunningException(
                                                "Task is not known or in state running on the JobManager."));

        stateWithExecutionGraph.deliverOperatorEventToCoordinator(taskExecution, operator, evt);
    }
@Test
    void testDeliverOperatorEventToCoordinatorFailsInIllegalState() throws Exception {
        final AdaptiveScheduler scheduler =
                new AdaptiveSchedulerBuilder(
                                createJobGraph(),
                                mainThreadExecutor,
                                EXECUTOR_RESOURCE.getExecutor())
                        .build();

        assertThatThrownBy(
                        () ->
                                scheduler.deliverOperatorEventToCoordinator(
                                        createExecutionAttemptId(),
                                        new OperatorID(),
                                        new TestOperatorEvent()))
                .isInstanceOf(TaskNotRunningException.class);
    }