@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;
  }
@ 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 { FlowGraphPath flowGraphPath = flowGraph . findPath ( flowSpec ) ; 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 ; }
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.