@Test()
public void testUnboundedSourceVertexDataFetching() throws Exception {

  final IRVertex sourceIRVertex = new TestUnboundedSourceVertex();
  final Long watermark = 1234567L;
  final BlockingQueue<Long> watermarkQueue = new LinkedBlockingQueue<>();
  watermarkQueue.add(watermark);
  final Readable readable = new TestUnboundedSourceReadable(watermarkQueue, 1);
  final Map<String, Readable> vertexIdToReadable = new HashMap<>();
  vertexIdToReadable.put(sourceIRVertex.getId(), readable);
  final List<Watermark> emittedWatermarks = new LinkedList<>();
  final Transform transform = new StreamTransformNoWatermarkEmit(emittedWatermarks);
  final OperatorVertex operatorVertex = new OperatorVertex(transform);
  final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag =
    new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>()
      .addVertex(sourceIRVertex)
      .addVertex(operatorVertex)
      .connectVertices(createEdge(sourceIRVertex, operatorVertex, "edge1"))
      .buildWithoutSourceSinkCheck();
  final StageEdge taskOutEdge = mockStageEdgeFrom(operatorVertex);
  final Task task =
    new Task(
      "testSourceVertexDataFetching",
      generateTaskId(),
      TASK_EXECUTION_PROPERTY_MAP,
      new byte[0],
      Collections.emptyList(),
      Collections.singletonList(taskOutEdge),
      vertexIdToReadable);
  // Execute the task.
  final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag);
  taskExecutor.execute();
  // Check whether the watermark is emitted
  assertEquals(Arrays.asList(new Watermark(watermark)), emittedWatermarks);
  // Check the output.
  assertEquals(elements, runtimeEdgeToOutputData.get(taskOutEdge.getId()));
}