code: private Dag<JobExecutionPlan> convertHopToDag(FlowEdgeContext flowEdgeContext, Config sysConfig)
        throws SpecNotFoundException, JobTemplate.TemplateException, URISyntaxException {
      FlowTemplate flowTemplate = flowEdgeContext.getEdge().getFlowTemplate();
      DatasetDescriptor inputDatasetDescriptor = flowEdgeContext.getInputDatasetDescriptor();
      DatasetDescriptor outputDatasetDescriptor = flowEdgeContext.getOutputDatasetDescriptor();
      Config mergedConfig = flowEdgeContext.getMergedConfig();
      SpecExecutor specExecutor = flowEdgeContext.getSpecExecutor();
  
      List<JobExecutionPlan> jobExecutionPlans = new ArrayList<>();
      Map<String, String> templateToJobNameMap = new HashMap<>();
  
      //Get resolved job configs from the flow template
      List<Config> resolvedJobConfigs = flowTemplate.getResolvedJobConfigs(mergedConfig, inputDatasetDescriptor, outputDatasetDescriptor);
      //Iterate over each resolved job config and convert the config to a JobSpec.
      for (Config resolvedJobConfig : resolvedJobConfigs) {
        JobExecutionPlan jobExecutionPlan = new JobExecutionPlan.Factory().createPlan(flowSpec, resolvedJobConfig, specExecutor, flowExecutionId, sysConfig);
        jobExecutionPlans.add(jobExecutionPlan);
        templateToJobNameMap.put(getJobTemplateName(jobExecutionPlan), jobExecutionPlan.getJobSpec().getConfig().getString(
            ConfigurationKeys.JOB_NAME_KEY));
      }
      updateJobDependencies(jobExecutionPlans, templateToJobNameMap);
      return new JobExecutionPlanDagFactory().createDag(jobExecutionPlans);
    }
docs: Given an instance of {@link FlowEdge}, this method returns a {@link Dag < JobExecutionPlan >} that moves data
  from the source of the {@link FlowEdge} to the destination of the {@link FlowEdge}.
  @param flowEdgeContext an instance of {@link FlowEdgeContext}.
  @param sysConfig environment config.
  @return a {@link Dag} of {@link JobExecutionPlan}s associated with the {@link FlowEdge}.