@Test public void testGetConsumedResultPartitionsProducers() throws Exception { final JobVertex producer1 = ExecutionGraphTestUtils.createNoOpVertex(1); final JobVertex producer2 = ExecutionGraphTestUtils.createNoOpVertex(1); final JobVertex consumer = ExecutionGraphTestUtils.createNoOpVertex(1); consumer.connectNewDataSetAsInput(producer1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED); consumer.connectNewDataSetAsInput(producer2, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED); final ExecutionGraph eg = ExecutionGraphTestUtils.createSimpleTestGraph(new JobID(), producer1, producer2, consumer); final ExecutionGraphToInputsLocationsRetrieverAdapter inputsLocationsRetriever = new ExecutionGraphToInputsLocationsRetrieverAdapter(eg); ExecutionVertexID evIdOfProducer1 = new ExecutionVertexID(producer1.getID(), 0); ExecutionVertexID evIdOfProducer2 = new ExecutionVertexID(producer2.getID(), 0); ExecutionVertexID evIdOfConsumer = new ExecutionVertexID(consumer.getID(), 0); Collection<Collection<ExecutionVertexID>> producersOfProducer1 = inputsLocationsRetriever.getConsumedResultPartitionsProducers(evIdOfProducer1); Collection<Collection<ExecutionVertexID>> producersOfProducer2 = inputsLocationsRetriever.getConsumedResultPartitionsProducers(evIdOfProducer2); Collection<Collection<ExecutionVertexID>> producersOfConsumer = inputsLocationsRetriever.getConsumedResultPartitionsProducers(evIdOfConsumer); assertThat(producersOfProducer1, is(empty())); assertThat(producersOfProducer2, is(empty())); assertThat(producersOfConsumer, hasSize(2)); assertThat(producersOfConsumer, hasItem(Collections.singletonList(evIdOfProducer1))); assertThat(producersOfConsumer, hasItem(Collections.singletonList(evIdOfProducer2))); }
@Override public Collection<Collection<ExecutionVertexID>> getConsumedResultPartitionsProducers( ExecutionVertexID executionVertexId) { ExecutionVertex ev = getExecutionVertex(executionVertexId); List<Collection<ExecutionVertexID>> resultPartitionProducers = new ArrayList<>(ev.getNumberOfInputs()); for (int i = 0; i < ev.getNumberOfInputs(); i++) { ExecutionEdge[] inputEdges = ev.getInputEdges(i); List<ExecutionVertexID> producers = new ArrayList<>(inputEdges.length); for (ExecutionEdge inputEdge : inputEdges) { ExecutionVertex producer = inputEdge.getSource().getProducer(); producers.add(new ExecutionVertexID(producer.getJobvertexId(), producer.getParallelSubtaskIndex())); } resultPartitionProducers.add(producers); } return resultPartitionProducers; }
ExecutionGraphToInputsLocationsRetrieverAdapter implements InputsLocationsRetriever { @Override public Collection<Collection<ExecutionVertexID>> getConsumedResultPartitionsProducers( ExecutionVertexID executionVertexId) { ExecutionVertex ev = getExecutionVertex(executionVertexId); List<Collection<ExecutionVertexID>> resultPartitionProducers = new ArrayList<>(ev.getNumberOfInputs()); for (int i = 0; i < ev.getNumberOfInputs(); i++) { ExecutionEdge[] inputEdges = ev.getInputEdges(i); List<ExecutionVertexID> producers = new ArrayList<>(inputEdges.length); for (ExecutionEdge inputEdge : inputEdges) { ExecutionVertex producer = inputEdge.getSource().getProducer(); producers.add(new ExecutionVertexID(producer.getJobvertexId(), producer.getParallelSubtaskIndex())); } resultPartitionProducers.add(producers); } return resultPartitionProducers; } }
ExecutionGraphToInputsLocationsRetrieverAdapter implements InputsLocationsRetriever { @Override public Collection<Collection<ExecutionVertexID>> getConsumedResultPartitionsProducers( ExecutionVertexID executionVertexId) { ExecutionVertex ev = getExecutionVertex(executionVertexId); List<Collection<ExecutionVertexID>> resultPartitionProducers = new ArrayList<>(ev.getNumberOfInputs()); for (int i = 0; i < ev.getNumberOfInputs(); i++) { ExecutionEdge[] inputEdges = ev.getInputEdges(i); List<ExecutionVertexID> producers = new ArrayList<>(inputEdges.length); for (ExecutionEdge inputEdge : inputEdges) { ExecutionVertex producer = inputEdge.getSource().getProducer(); producers.add(new ExecutionVertexID(producer.getJobvertexId(), producer.getParallelSubtaskIndex())); } resultPartitionProducers.add(producers); } return resultPartitionProducers; } ExecutionGraphToInputsLocationsRetrieverAdapter(ExecutionGraph executionGraph); }
ExecutionGraphToInputsLocationsRetrieverAdapter implements InputsLocationsRetriever { @Override public Collection<Collection<ExecutionVertexID>> getConsumedResultPartitionsProducers( ExecutionVertexID executionVertexId) { ExecutionVertex ev = getExecutionVertex(executionVertexId); List<Collection<ExecutionVertexID>> resultPartitionProducers = new ArrayList<>(ev.getNumberOfInputs()); for (int i = 0; i < ev.getNumberOfInputs(); i++) { ExecutionEdge[] inputEdges = ev.getInputEdges(i); List<ExecutionVertexID> producers = new ArrayList<>(inputEdges.length); for (ExecutionEdge inputEdge : inputEdges) { ExecutionVertex producer = inputEdge.getSource().getProducer(); producers.add(new ExecutionVertexID(producer.getJobvertexId(), producer.getParallelSubtaskIndex())); } resultPartitionProducers.add(producers); } return resultPartitionProducers; } ExecutionGraphToInputsLocationsRetrieverAdapter(ExecutionGraph executionGraph); @Override Collection<Collection<ExecutionVertexID>> getConsumedResultPartitionsProducers(
			ExecutionVertexID executionVertexId); @Override Optional<CompletableFuture<TaskManagerLocation>> getTaskManagerLocation(ExecutionVertexID executionVertexId); }
ExecutionGraphToInputsLocationsRetrieverAdapter implements InputsLocationsRetriever { @Override public Collection<Collection<ExecutionVertexID>> getConsumedResultPartitionsProducers( ExecutionVertexID executionVertexId) { ExecutionVertex ev = getExecutionVertex(executionVertexId); List<Collection<ExecutionVertexID>> resultPartitionProducers = new ArrayList<>(ev.getNumberOfInputs()); for (int i = 0; i < ev.getNumberOfInputs(); i++) { ExecutionEdge[] inputEdges = ev.getInputEdges(i); List<ExecutionVertexID> producers = new ArrayList<>(inputEdges.length); for (ExecutionEdge inputEdge : inputEdges) { ExecutionVertex producer = inputEdge.getSource().getProducer(); producers.add(new ExecutionVertexID(producer.getJobvertexId(), producer.getParallelSubtaskIndex())); } resultPartitionProducers.add(producers); } return resultPartitionProducers; } ExecutionGraphToInputsLocationsRetrieverAdapter(ExecutionGraph executionGraph); @Override Collection<Collection<ExecutionVertexID>> getConsumedResultPartitionsProducers(
			ExecutionVertexID executionVertexId); @Override Optional<CompletableFuture<TaskManagerLocation>> getTaskManagerLocation(ExecutionVertexID executionVertexId);  }