gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
/**
     * Main body of the {@link DagManagerThread}. Deque the next item from the queue and poll job statuses of currently
     * running jobs.
     */
Main body of the  DagManagerThread. Deque the next item from the queue and poll job statuses of currently
running jobs.
Main body of the DagManagerThread . Deque the next item from the queue and poll job statuses of currently running jobs .
@Override
    public void run() {
      try {
        String nextDagToCancel = cancelQueue.poll();
        if (nextDagToCancel != null) {
          cancelDag(nextDagToCancel);
        }
        Dag<JobExecutionPlan> dag = queue.poll();
        if (dag != null) {
          if (dag.isEmpty()) {
            log.info("Empty dag; ignoring the dag");
          }
          initialize(dag);
        }
        log.debug("Polling job statuses..");
        pollAndAdvanceDag();
        log.debug("Poll done.");
        log.debug("Cleaning up finished dags..");
        cleanUp();
        log.debug("Clean up done");
      } catch (Exception e) {
        log.error("Exception encountered in {}", getClass().getName(), e);
      }
    }
@ Override public void run ( ) { try { String nextDagToCancel = cancelQueue . poll ( ) ; if ( nextDagToCancel != null ) { cancelDag ( nextDagToCancel ) ; } Dag < JobExecutionPlan > dag = queue . poll ( ) ; if ( dag != null ) { if ( dag . isEmpty ( ) ) { log . info ( "Empty dag; ignoring the dag" ) ; } initialize ( dag ) ; } log . debug ( "Polling job statuses.." ) ; pollAndAdvanceDag ( ) ; log . debug ( "Poll done." ) ; log . debug ( "Cleaning up finished dags.." ) ; cleanUp ( ) ; log . debug ( "Clean up done" ) ; } catch ( Exception e ) { log . error ( "Exception encountered in {}" , getClass ( ) . getName ( ) , e ) ; } }
Main body of the {@link DagManagerThread}.
Main body of the { @link DagManagerThread } .
//Poll the cancelQueue for a new Dag to cancel. //Poll the queue for a new Dag to execute. //Initialize dag. //Poll and update the job statuses of running jobs. //Clean up any finished dags