@Override
    public ExecutionState requestPartitionState(
            IntermediateDataSetID intermediateResultId, ResultPartitionID resultPartitionId)
            throws PartitionProducerDisposedException {
        return state.tryCall(
                        StateWithExecutionGraph.class,
                        stateWithExecutionGraph ->
                                stateWithExecutionGraph.requestPartitionState(
                                        intermediateResultId, resultPartitionId),
                        "requestPartitionState")
                .orElseThrow(() -> new PartitionProducerDisposedException(resultPartitionId));
    }
@Test
    void testRequestPartitionStateFailsInIllegalState() throws Exception {
        final AdaptiveScheduler scheduler =
                new AdaptiveSchedulerBuilder(
                                createJobGraph(),
                                mainThreadExecutor,
                                EXECUTOR_RESOURCE.getExecutor())
                        .build();

        assertThatThrownBy(
                        () ->
                                scheduler.requestPartitionState(
                                        new IntermediateDataSetID(), new ResultPartitionID()))
                .isInstanceOf(PartitionProducerDisposedException.class);
    }