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

  final List<Watermark> emittedWatermarks = new ArrayList<>();
  final IRVertex operatorIRVertex1 = new OperatorVertex(new StreamTransform());
  final IRVertex operatorIRVertex2 = new OperatorVertex(new StreamTransformNoWatermarkEmit(emittedWatermarks));
  final IRVertex operatorIRVertex3 = new OperatorVertex(new StreamTransform());
  final IRVertex sourceIRVertex1 = new TestUnboundedSourceVertex();
  final IRVertex sourceIRVertex2 = new TestUnboundedSourceVertex();
  final Queue<Long> watermarks1 = new ConcurrentLinkedQueue<>();
  watermarks1.add(500L);
  final Queue<Long> watermarks2 = new ConcurrentLinkedQueue<>();
  watermarks2.add(1000L);
  final Readable readable1 = new TestUnboundedSourceReadable(watermarks1, 5);
  final Readable readable2 = new TestUnboundedSourceReadable(watermarks2, 2);
  final Map<String, Readable> vertexIdToReadable = new HashMap<>();
  vertexIdToReadable.put(sourceIRVertex1.getId(), readable1);
  vertexIdToReadable.put(sourceIRVertex2.getId(), readable2);
  final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag =
    new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>()
      .addVertex(sourceIRVertex1)
      .addVertex(sourceIRVertex2)
      .addVertex(operatorIRVertex1)
      .addVertex(operatorIRVertex2)
      .addVertex(operatorIRVertex3)
      .connectVertices(createEdge(sourceIRVertex1, operatorIRVertex1, "edge1"))
      .connectVertices(createEdge(operatorIRVertex1, operatorIRVertex2, "edge2"))
      .connectVertices(createEdge(sourceIRVertex2, operatorIRVertex3, "edge3"))
      .connectVertices(createEdge(operatorIRVertex3, operatorIRVertex2, "edge4"))
      .buildWithoutSourceSinkCheck();
  final StageEdge taskOutEdge = mockStageEdgeFrom(operatorIRVertex2);
  final Task task =
    new Task(
      "testSourceVertexDataFetching",
      generateTaskId(),
      TASK_EXECUTION_PROPERTY_MAP,
      new byte[0],
      Collections.emptyList(),
      Collections.singletonList(taskOutEdge),
      vertexIdToReadable);
  // Execute the task.
  final Thread watermarkEmitThread = new Thread(new Runnable() {
    @Override
    public void run() {
      waitUntilWatermarkEmitted(watermarks2);
      watermarks1.add(600L);
      watermarks1.add(1400L);
      watermarks1.add(1800L);
      waitUntilWatermarkEmitted(watermarks1);
      watermarks2.add(2200L);
      waitUntilWatermarkEmitted(watermarks2);
      watermarks1.add(2500L);
      waitUntilWatermarkEmitted(watermarks1);
    }
  });
  watermarkEmitThread.start();
  final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag);
  taskExecutor.execute();
  watermarkEmitThread.join();
  // Check whether the watermark is emitted
  assertEquals(Arrays.asList(
    new Watermark(500), new Watermark(600), new Watermark(1000),
    new Watermark(1800), new Watermark(2200)), emittedWatermarks);
  // Check the output.
  final List<Integer> doubledElements = new ArrayList<>(elements.size() * 2);
  doubledElements.addAll(elements);
  doubledElements.addAll(elements);
  assertTrue(checkEqualElements(doubledElements, runtimeEdgeToOutputData.get(taskOutEdge.getId())));
}