@Override
    public Optional<CompletableFuture<TaskManagerLocation>> getTaskManagerLocation(
            ExecutionVertexID executionVertexId) {
        return inputsLocationsRetriever
                .getTaskManagerLocation(executionVertexId)
                .filter(future -> future.isDone() && !future.isCompletedExceptionally());
    }
@Test
    void testNoInputLocationIfNotDone() {
        TestingInputsLocationsRetriever originalLocationRetriever = getOriginalLocationRetriever();
        originalLocationRetriever.markScheduled(EV1);
        InputsLocationsRetriever availableInputsLocationsRetriever =
                new AvailableInputsLocationsRetriever(originalLocationRetriever);
        assertThat(availableInputsLocationsRetriever.getTaskManagerLocation(EV1)).isNotPresent();
    }