@Override
    public SerializedInputSplit requestNextInputSplit(
            JobVertexID vertexID, ExecutionAttemptID executionAttempt) throws IOException {
        return state.tryCall(
                        StateWithExecutionGraph.class,
                        stateWithExecutionGraph ->
                                stateWithExecutionGraph.requestNextInputSplit(
                                        vertexID, executionAttempt),
                        "requestNextInputSplit")
                .orElseThrow(
                        () -> new IOException("Scheduler is currently not executing the job."));
    }
@Test
    void testRequestNextInputSplitFailsInIllegalState() throws Exception {
        final AdaptiveScheduler scheduler =
                new AdaptiveSchedulerBuilder(
                                createJobGraph(),
                                mainThreadExecutor,
                                EXECUTOR_RESOURCE.getExecutor())
                        .build();

        assertThatThrownBy(
                        () ->
                                scheduler.requestNextInputSplit(
                                        JOB_VERTEX.getID(), createExecutionAttemptId()))
                .isInstanceOf(IOException.class);
    }