@Override
    public Optional<CompletableFuture<TaskManagerLocation>> getTaskManagerLocation(
            ExecutionVertexID executionVertexId) {
        ExecutionVertex ev = getExecutionVertex(executionVertexId);

        if (ev.getExecutionState() != ExecutionState.CREATED) {
            return Optional.of(ev.getCurrentTaskManagerLocationFuture());
        } else {
            return Optional.empty();
        }
    }
@Test
    void testGetTaskManagerLocationWhenScheduled() throws Exception {
        final JobVertex jobVertex = ExecutionGraphTestUtils.createNoOpVertex(1);

        final TestingLogicalSlot testingLogicalSlot =
                new TestingLogicalSlotBuilder().createTestingLogicalSlot();
        final ExecutionGraph eg =
                ExecutionGraphTestUtils.createExecutionGraph(
                        EXECUTOR_EXTENSION.getExecutor(), jobVertex);
        final ExecutionGraphToInputsLocationsRetrieverAdapter inputsLocationsRetriever =
                new ExecutionGraphToInputsLocationsRetrieverAdapter(eg);

        final ExecutionVertex onlyExecutionVertex = eg.getAllExecutionVertices().iterator().next();
        onlyExecutionVertex.getCurrentExecutionAttempt().transitionState(ExecutionState.SCHEDULED);
        onlyExecutionVertex.deployToSlot(testingLogicalSlot);

        ExecutionVertexID executionVertexId = new ExecutionVertexID(jobVertex.getID(), 0);
        Optional<CompletableFuture<TaskManagerLocation>> taskManagerLocationOptional =
                inputsLocationsRetriever.getTaskManagerLocation(executionVertexId);

        assertThat(taskManagerLocationOptional).isPresent();

        final CompletableFuture<TaskManagerLocation> taskManagerLocationFuture =
                taskManagerLocationOptional.get();
        assertThat(taskManagerLocationFuture.get())
                .isEqualTo(testingLogicalSlot.getTaskManagerLocation());
    }