gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
/**
     * This method determines the next set of jobs to execute from the dag and submits them for execution.
     * This method updates internal data structures tracking currently running Dags and jobs.
     */
This method determines the next set of jobs to execute from the dag and submits them for execution.
This method updates internal data structures tracking currently running Dags and jobs.
This method determines the next set of jobs to execute from the dag and submits them for execution . This method updates internal data structures tracking currently running Dags and jobs .
private void initialize(Dag<JobExecutionPlan> dag)
        throws IOException {
      String dagId = DagManagerUtils.generateDagId(dag);
      log.info("Initializing Dag {}", DagManagerUtils.getFullyQualifiedDagName(dag));
      if (this.dags.containsKey(dagId)) {
        log.warn("Already tracking a dag with dagId {}, skipping.", dagId);
        return;
      }
      this.dags.put(dagId, dag);
      log.debug("Dag {} - determining if any jobs are already running.", DagManagerUtils.getFullyQualifiedDagName(dag));
      for (DagNode<JobExecutionPlan> dagNode : dag.getNodes()) {
        if (DagManagerUtils.getExecutionStatus(dagNode) == RUNNING) {
          addJobState(dagId, dagNode);
        }
      }
      log.debug("Dag {} submitting jobs ready for execution.", DagManagerUtils.getFullyQualifiedDagName(dag));
      Map<String, Set<DagNode<JobExecutionPlan>>> nextSubmitted = submitNext(dagId);
      for (DagNode dagNode: nextSubmitted.get(dagId)) {
        addJobState(dagId, dagNode);
      }
      log.info("Dag {} Initialization complete.", DagManagerUtils.getFullyQualifiedDagName(dag));
    }
private void initialize ( Dag < JobExecutionPlan > dag ) throws IOException { String dagId = DagManagerUtils . generateDagId ( dag ) ; log . info ( "Initializing Dag {}" , DagManagerUtils . getFullyQualifiedDagName ( dag ) ) ; if ( this . dags . containsKey ( dagId ) ) { log . warn ( "Already tracking a dag with dagId {}, skipping." , dagId ) ; return ; } this . dags . put ( dagId , dag ) ; log . debug ( "Dag {} - determining if any jobs are already running." , DagManagerUtils . getFullyQualifiedDagName ( dag ) ) ; for ( DagNode < JobExecutionPlan > dagNode : dag . getNodes ( ) ) { if ( DagManagerUtils . getExecutionStatus ( dagNode ) == RUNNING ) { addJobState ( dagId , dagNode ) ; } } log . debug ( "Dag {} submitting jobs ready for execution." , DagManagerUtils . getFullyQualifiedDagName ( dag ) ) ; Map < String , Set < DagNode < JobExecutionPlan > > > nextSubmitted = submitNext ( dagId ) ; for ( DagNode dagNode : nextSubmitted . get ( dagId ) ) { addJobState ( dagId , dagNode ) ; } log . info ( "Dag {} Initialization complete." , DagManagerUtils . getFullyQualifiedDagName ( dag ) ) ; }
This method determines the next set of jobs to execute from the dag and submits them for execution.
This method determines the next set of jobs to execute from the dag and submits them for execution .
//Add Dag to the map of running dags //Are there any jobs already in the running state? This check is for Dags already running //before a leadership change occurs. //Determine the next set of jobs to run and submit them for execution