@Test public void testGetNonExistingExecutionVertexWillThrowException() throws Exception { final JobVertex jobVertex = ExecutionGraphTestUtils.createNoOpVertex(1); final ExecutionGraph eg = ExecutionGraphTestUtils.createSimpleTestGraph(new JobID(), jobVertex); final ExecutionGraphToInputsLocationsRetrieverAdapter inputsLocationsRetriever = new ExecutionGraphToInputsLocationsRetrieverAdapter(eg); ExecutionVertexID invalidExecutionVertexId = new ExecutionVertexID(new JobVertexID(), 0); try { inputsLocationsRetriever.getTaskManagerLocation(invalidExecutionVertexId); fail("Should throw exception if execution vertex doesn't exist!"); } catch (IllegalStateException expected) { } }
@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);  }