@Test public void testGetTaskManagerLocationWhenScheduled() throws Exception { final JobVertex jobVertex = ExecutionGraphTestUtils.createNoOpVertex(1); final TestingLogicalSlot testingLogicalSlot = new TestingLogicalSlotBuilder().createTestingLogicalSlot(); final ExecutionGraph eg = ExecutionGraphTestUtils.createSimpleTestGraph(new JobID(), jobVertex); final ExecutionGraphToInputsLocationsRetrieverAdapter inputsLocationsRetriever = new ExecutionGraphToInputsLocationsRetrieverAdapter(eg); final ExecutionVertex onlyExecutionVertex = eg.getAllExecutionVertices().iterator().next(); onlyExecutionVertex.deployToSlot(testingLogicalSlot); ExecutionVertexID executionVertexId = new ExecutionVertexID(jobVertex.getID(), 0); Optional<CompletableFuture<TaskManagerLocation>> taskManagerLocationOptional = inputsLocationsRetriever.getTaskManagerLocation(executionVertexId); assertTrue(taskManagerLocationOptional.isPresent()); final CompletableFuture<TaskManagerLocation> taskManagerLocationFuture = taskManagerLocationOptional.get(); assertThat(taskManagerLocationFuture.get(), is(testingLogicalSlot.getTaskManagerLocation())); }
@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);  }