@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 testGetNonExistingExecutionVertexWillThrowException() throws Exception {
        final JobVertex jobVertex = ExecutionGraphTestUtils.createNoOpVertex(1);

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

        ExecutionVertexID invalidExecutionVertexId = new ExecutionVertexID(new JobVertexID(), 0);
        assertThatThrownBy(
                        () ->
                                inputsLocationsRetriever.getTaskManagerLocation(
                                        invalidExecutionVertexId),
                        "Should throw exception if execution vertex doesn't exist!")
                .isInstanceOf(IllegalStateException.class);
    }