  @Test (dependsOnMethods = "testCompileFlowSingleHop")
  public void testMulticastPath() throws IOException, URISyntaxException {

    FlowSpec spec = createFlowSpec("flow/flow2.conf", "LocalFS-1", "HDFS-3,HDFS-4", false, false);
    Dag<JobExecutionPlan> jobDag = this.specCompiler.compileFlow(spec);

    Assert.assertEquals(jobDag.getNodes().size(), 4);
    Assert.assertEquals(jobDag.getEndNodes().size(), 2);
    Assert.assertEquals(jobDag.getStartNodes().size(), 2);

    //First hop must be from LocalFS to HDFS-1 and HDFS-2
    Set<String> jobNames = new HashSet<>();
    jobNames.add(Joiner.on(JobExecutionPlan.Factory.JOB_NAME_COMPONENT_SEPARATION_CHAR).
        join("testFlowGroup", "testFlowName", "Distcp", "LocalFS-1", "HDFS-1", "localToHdfs"));
    jobNames.add(Joiner.on(JobExecutionPlan.Factory.JOB_NAME_COMPONENT_SEPARATION_CHAR).
        join("testFlowGroup", "testFlowName", "Distcp", "LocalFS-1", "HDFS-2", "localToHdfs"));

    for (DagNode<JobExecutionPlan> dagNode : jobDag.getStartNodes()) {
      Config jobConfig = dagNode.getValue().getJobSpec().getConfig();
      String jobName = jobConfig.getString(ConfigurationKeys.JOB_NAME_KEY);
      Assert.assertTrue(jobNames.stream().anyMatch(jobName::startsWith));
    }

    //Second hop must be from HDFS-1/HDFS-2 to HDFS-3/HDFS-4 respectively.
    jobNames = new HashSet<>();
    jobNames.add(Joiner.on(JobExecutionPlan.Factory.JOB_NAME_COMPONENT_SEPARATION_CHAR).
        join("testFlowGroup", "testFlowName", "Distcp", "HDFS-1", "HDFS-3", "hdfsToHdfs"));
    jobNames.add(Joiner.on(JobExecutionPlan.Factory.JOB_NAME_COMPONENT_SEPARATION_CHAR).
        join("testFlowGroup", "testFlowName", "Distcp", "HDFS-2", "HDFS-4", "hdfsToHdfs"));
    for (DagNode<JobExecutionPlan> dagNode : jobDag.getStartNodes()) {
      List<DagNode<JobExecutionPlan>> nextNodes = jobDag.getChildren(dagNode);
      Assert.assertEquals(nextNodes.size(), 1);
      Config jobConfig = nextNodes.get(0).getValue().getJobSpec().getConfig();
      String jobName = jobConfig.getString(ConfigurationKeys.JOB_NAME_KEY);
      Assert.assertTrue(jobNames.stream().anyMatch(jobName::startsWith));
    }
  }