code: @Override
    public Dag<JobExecutionPlan> compileFlow(Spec spec) {
      Preconditions.checkNotNull(spec);
      Preconditions.checkArgument(spec instanceof FlowSpec, "MultiHopFlowCompiler only accepts FlowSpecs");
  
      long startTime = System.nanoTime();
  
      FlowSpec flowSpec = (FlowSpec) spec;
      String source = ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY, "");
      String destination =
          ConfigUtils.getString(flowSpec.getConfig(), ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY, "");
      log.info(String.format("Compiling flow for source: %s and destination: %s", source, destination));
  
      Dag<JobExecutionPlan> jobExecutionPlanDag;
      try {
        //Compute the path from source to destination.
        FlowGraphPath flowGraphPath = flowGraph.findPath(flowSpec);
        //Convert the path into a Dag of JobExecutionPlans.
        if (flowGraphPath != null) {
          jobExecutionPlanDag = flowGraphPath.asDag(this.config);
        } else {
          Instrumented.markMeter(flowCompilationFailedMeter);
          log.info(String.format("No path found from source: %s and destination: %s", source, destination));
          return new JobExecutionPlanDagFactory().createDag(new ArrayList<>());
        }
      } catch (PathFinder.PathFinderException | SpecNotFoundException | JobTemplate.TemplateException | URISyntaxException | ReflectiveOperationException e) {
        Instrumented.markMeter(flowCompilationFailedMeter);
        log.error(String
                .format("Exception encountered while compiling flow for source: %s and destination: %s", source, destination),
            e);
        return null;
      }
      Instrumented.markMeter(flowCompilationSuccessFulMeter);
      Instrumented.updateTimer(flowCompilationTimer, System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
      return jobExecutionPlanDag;
    }
docs: j
  @param spec an instance of {@link FlowSpec}.
  @return A DAG of {@link JobExecutionPlan}s, which encapsulates the compiled {@link org.apache.gobblin.runtime.api.JobSpec}s
  together with the {@link SpecExecutor} where the job can be executed.