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

  final String additionalTag1 = "bonus1";
  final String additionalTag2 = "bonus2";
  final IRVertex routerVertex = new OperatorVertex(
    new RoutingTransform(Arrays.asList(additionalTag1, additionalTag2)));
  final IRVertex mainVertex = new OperatorVertex(new StreamTransform());
  final IRVertex bonusVertex1 = new OperatorVertex(new StreamTransform());
  final IRVertex bonusVertex2 = new OperatorVertex(new StreamTransform());
  final RuntimeEdge<IRVertex> edge1 = createEdge(routerVertex, mainVertex, "edge-1");
  final RuntimeEdge<IRVertex> edge2 = createEdge(routerVertex, bonusVertex1, "edge-2");
  final RuntimeEdge<IRVertex> edge3 = createEdge(routerVertex, bonusVertex2, "edge-3");
  edge2.getExecutionProperties().put(AdditionalOutputTagProperty.of(additionalTag1));
  edge3.getExecutionProperties().put(AdditionalOutputTagProperty.of(additionalTag2));
  final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag = new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>()
    .addVertex(routerVertex)
    .addVertex(mainVertex)
    .addVertex(bonusVertex1)
    .addVertex(bonusVertex2)
    .connectVertices(edge1)
    .connectVertices(edge2)
    .connectVertices(edge3)
    .buildWithoutSourceSinkCheck();
  final StageEdge outEdge1 = mockStageEdgeFrom(mainVertex);
  final StageEdge outEdge2 = mockStageEdgeFrom(bonusVertex1);
  final StageEdge outEdge3 = mockStageEdgeFrom(bonusVertex2);
  final Task task = new Task(
    "testAdditionalOutputs",
    generateTaskId(),
    TASK_EXECUTION_PROPERTY_MAP,
    new byte[0],
    Collections.singletonList(mockStageEdgeTo(routerVertex)),
    Arrays.asList(outEdge1, outEdge2, outEdge3),
    Collections.emptyMap());
  // Execute the task.
  final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag);
  taskExecutor.execute();
  // Check the output.
  final List<Integer> mainOutputs = runtimeEdgeToOutputData.get(outEdge1.getId());
  final List<Integer> bonusOutputs1 = runtimeEdgeToOutputData.get(outEdge2.getId());
  final List<Integer> bonusOutputs2 = runtimeEdgeToOutputData.get(outEdge3.getId());
  List<Integer> even = elements.stream().filter(i -> i % 2 == 0).collect(Collectors.toList());
  List<Integer> odd = elements.stream().filter(i -> i % 2 != 0).collect(Collectors.toList());
  assertTrue(checkEqualElements(even, mainOutputs));
  assertTrue(checkEqualElements(odd, bonusOutputs1));
  assertTrue(checkEqualElements(odd, bonusOutputs2));
}