@Test public void testGetEmptyTaskManagerLocationIfVertexNotScheduled() throws Exception { final JobVertex jobVertex = ExecutionGraphTestUtils.createNoOpVertex(1); final ExecutionGraph eg = ExecutionGraphTestUtils.createSimpleTestGraph(new JobID(), jobVertex); final ExecutionGraphToInputsLocationsRetrieverAdapter inputsLocationsRetriever = new ExecutionGraphToInputsLocationsRetrieverAdapter(eg); ExecutionVertexID executionVertexId = new ExecutionVertexID(jobVertex.getID(), 0); Optional<CompletableFuture<TaskManagerLocation>> taskManagerLocation = inputsLocationsRetriever.getTaskManagerLocation(executionVertexId); assertFalse(taskManagerLocation.isPresent()); }
@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(); } }
ExecutionGraphToInputsLocationsRetrieverAdapter implements InputsLocationsRetriever { @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(); } } }
ExecutionGraphToInputsLocationsRetrieverAdapter implements InputsLocationsRetriever { @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(); } } ExecutionGraphToInputsLocationsRetrieverAdapter(ExecutionGraph executionGraph); }
ExecutionGraphToInputsLocationsRetrieverAdapter implements InputsLocationsRetriever { @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(); } } ExecutionGraphToInputsLocationsRetrieverAdapter(ExecutionGraph executionGraph); @Override Collection<Collection<ExecutionVertexID>> getConsumedResultPartitionsProducers(
			ExecutionVertexID executionVertexId); @Override Optional<CompletableFuture<TaskManagerLocation>> getTaskManagerLocation(ExecutionVertexID executionVertexId); }
ExecutionGraphToInputsLocationsRetrieverAdapter implements InputsLocationsRetriever { @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(); } } ExecutionGraphToInputsLocationsRetrieverAdapter(ExecutionGraph executionGraph); @Override Collection<Collection<ExecutionVertexID>> getConsumedResultPartitionsProducers(
			ExecutionVertexID executionVertexId); @Override Optional<CompletableFuture<TaskManagerLocation>> getTaskManagerLocation(ExecutionVertexID executionVertexId);  }