@Test(timeout = 5000)
public void testParentTaskDataFetching() throws Exception {

  final IRVertex vertex = new OperatorVertex(new StreamTransform());
  final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag = new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>()
    .addVertex(vertex)
    .buildWithoutSourceSinkCheck();
  final StageEdge taskOutEdge = mockStageEdgeFrom(vertex);
  final Task task = new Task(
    "testSourceVertexDataFetching",
    generateTaskId(),
    TASK_EXECUTION_PROPERTY_MAP,
    new byte[0],
    Collections.singletonList(mockStageEdgeTo(vertex)),
    Collections.singletonList(taskOutEdge),
    Collections.emptyMap());
  // Execute the task.
  final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag);
  taskExecutor.execute();
  // Check the output.
  assertTrue(checkEqualElements(elements, runtimeEdgeToOutputData.get(taskOutEdge.getId())));
}