@Override
    public Iterable<DefaultLogicalResult> getConsumedResults() {
        return jobVertex.getInputs().stream()
                .map(JobEdge::getSource)
                .map(IntermediateDataSet::getId)
                .map(resultRetriever)
                .collect(Collectors.toList());
    }
@Test
    public void testGetConsumedResults() {
        assertResultsEquals(results, downstreamLogicalVertex.getConsumedResults());
    }