code: @VisibleForTesting
    static Dag<JobExecutionPlan> concatenate(Dag<JobExecutionPlan> dagLeft, Dag<JobExecutionPlan> dagRight) {
      //Compute the fork nodes - set of nodes with no dependents in the concatenated dag.
      Set<DagNode<JobExecutionPlan>> forkNodes = dagLeft.getEndNodes().stream().
          filter(endNode -> isNodeForkable(endNode)).collect(Collectors.toSet());
      Set<DagNode<JobExecutionPlan>> dependencyNodes = dagLeft.getDependencyNodes(forkNodes);
  
      if (!dependencyNodes.isEmpty()) {
        List<String> dependenciesList = dependencyNodes.stream()
            .map(dagNode -> dagNode.getValue().getJobSpec().getConfig().getString(ConfigurationKeys.JOB_NAME_KEY))
            .collect(Collectors.toList());
        String dependencies = Joiner.on(",").join(dependenciesList);
        for (DagNode<JobExecutionPlan> childNode : dagRight.getStartNodes()) {
          JobSpec jobSpec = childNode.getValue().getJobSpec();
          jobSpec.setConfig(jobSpec.getConfig().withValue(ConfigurationKeys.JOB_DEPENDENCIES,
              ConfigValueFactory.fromAnyRef(dependencies)));
        }
      }
  
      return dagLeft.concatenate(dagRight, forkNodes);
    }
docs: Concatenate two {@link Dag}s. Modify the {@link ConfigurationKeys#JOB_DEPENDENCIES} in the {@link JobSpec}s of the child
  {@link Dag} to reflect the concatenation operation.
  @param dagLeft The parent dag.
  @param dagRight The child dag.
  @return The concatenated dag with modified {@link ConfigurationKeys#JOB_DEPENDENCIES}.