@Test(timeout = 5000)
public void testTwoOperatorsWithBroadcastVariable() {

  final Transform singleListTransform = new CreateSingleListTransform();
  final long broadcastId = 0;
  final IRVertex operatorIRVertex1 = new OperatorVertex(new StreamTransform());
  final IRVertex operatorIRVertex2 = new OperatorVertex(new BroadcastVariablePairingTransform(broadcastId));
  final String edgeId = "edge";
  final DAG<IRVertex, RuntimeEdge<IRVertex>> taskDag = new DAGBuilder<IRVertex, RuntimeEdge<IRVertex>>()
    .addVertex(operatorIRVertex1)
    .addVertex(operatorIRVertex2)
    .connectVertices(createEdge(operatorIRVertex1, operatorIRVertex2, edgeId))
    .buildWithoutSourceSinkCheck();
  final StageEdge taskOutEdge = mockStageEdgeFrom(operatorIRVertex2);
  final StageEdge taskInEdge = mockStageEdgeTo(operatorIRVertex1);
  when(broadcastManagerWorker.get(broadcastId)).thenReturn(new ArrayList<>(elements));
  final Task task = new Task(
    "testSourceVertexDataFetching",
    generateTaskId(),
    TASK_EXECUTION_PROPERTY_MAP,
    new byte[0],
    Collections.singletonList(taskInEdge),
    Collections.singletonList(taskOutEdge),
    Collections.emptyMap());
  // Execute the task.
  final TaskExecutor taskExecutor = getTaskExecutor(task, taskDag);
  taskExecutor.execute();
  // Check the output.
  final List<Pair<List<Integer>, Integer>> pairs = runtimeEdgeToOutputData.get(taskOutEdge.getId());
  final List<Integer> values = pairs.stream().map(Pair::right).collect(Collectors.toList());
  assertTrue(checkEqualElements(elements, values));
  assertTrue(pairs.stream().map(Pair::left).allMatch(broadcastVar -> checkEqualElements(broadcastVar, values)));
}