gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
/**
     * Proceed the execution of each dag node based on job status.
     */
Proceed the execution of each dag node based on job status.
Proceed the execution of each dag node based on job status .
private void pollAndAdvanceDag()
        throws IOException {
      this.failedDagIdsFinishRunning.clear();
      Map<String, Set<DagNode<JobExecutionPlan>>> nextSubmitted = Maps.newHashMap();
      List<DagNode<JobExecutionPlan>> nodesToCleanUp = Lists.newArrayList();
      for (DagNode<JobExecutionPlan> node: this.jobToDag.keySet()) {
        long pollStartTime = System.nanoTime();
        JobStatus jobStatus = pollJobStatus(node);
        Instrumented.updateTimer(this.jobStatusPolledTimer, System.nanoTime() - pollStartTime, TimeUnit.NANOSECONDS);
        if (jobStatus == null) {
          continue;
        }
        JobExecutionPlan jobExecutionPlan = DagManagerUtils.getJobExecutionPlan(node);
        ExecutionStatus status = valueOf(jobStatus.getEventName());
        switch (status) {
          case COMPLETE:
            jobExecutionPlan.setExecutionStatus(COMPLETE);
            nextSubmitted.putAll(onJobFinish(node));
            nodesToCleanUp.add(node);
            break;
          case FAILED:
          case CANCELLED:
            jobExecutionPlan.setExecutionStatus(FAILED);
            nextSubmitted.putAll(onJobFinish(node));
            nodesToCleanUp.add(node);
            break;
          case PENDING:
            jobExecutionPlan.setExecutionStatus(PENDING);
            break;
          default:
            jobExecutionPlan.setExecutionStatus(RUNNING);
            break;
        }
        if (jobStatus.isShouldRetry()) {
          log.info("Retrying job: {}, current attempts: {}, max attempts: {}", DagManagerUtils.getFullyQualifiedJobName(node),
              jobStatus.getCurrentAttempts(), jobStatus.getMaxAttempts());
          submitJob(node);
        }
      }
      for (Map.Entry<String, Set<DagNode<JobExecutionPlan>>> entry: nextSubmitted.entrySet()) {
        String dagId = entry.getKey();
        Set<DagNode<JobExecutionPlan>> dagNodes = entry.getValue();
        for (DagNode dagNode: dagNodes) {
          addJobState(dagId, dagNode);
        }
      }
      for (DagNode<JobExecutionPlan> dagNode: nodesToCleanUp) {
        String dagId = DagManagerUtils.generateDagId(this.jobToDag.get(dagNode));
        deleteJobState(dagId, dagNode);
      }
    }
private void pollAndAdvanceDag ( ) throws IOException { this . failedDagIdsFinishRunning . clear ( ) ; Map < String , Set < DagNode < JobExecutionPlan > > > nextSubmitted = Maps . newHashMap ( ) ; List < DagNode < JobExecutionPlan > > nodesToCleanUp = Lists . newArrayList ( ) ; for ( DagNode < JobExecutionPlan > node : this . jobToDag . keySet ( ) ) { long pollStartTime = System . nanoTime ( ) ; JobStatus jobStatus = pollJobStatus ( node ) ; Instrumented . updateTimer ( this . jobStatusPolledTimer , System . nanoTime ( ) - pollStartTime , TimeUnit . NANOSECONDS ) ; if ( jobStatus == null ) { continue ; } JobExecutionPlan jobExecutionPlan = DagManagerUtils . getJobExecutionPlan ( node ) ; ExecutionStatus status = valueOf ( jobStatus . getEventName ( ) ) ; switch ( status ) { case COMPLETE : jobExecutionPlan . setExecutionStatus ( COMPLETE ) ; nextSubmitted . putAll ( onJobFinish ( node ) ) ; nodesToCleanUp . add ( node ) ; break ; case FAILED : case CANCELLED : jobExecutionPlan . setExecutionStatus ( FAILED ) ; nextSubmitted . putAll ( onJobFinish ( node ) ) ; nodesToCleanUp . add ( node ) ; break ; case PENDING : jobExecutionPlan . setExecutionStatus ( PENDING ) ; break ; default : jobExecutionPlan . setExecutionStatus ( RUNNING ) ; break ; } if ( jobStatus . isShouldRetry ( ) ) { log . info ( "Retrying job: {}, current attempts: {}, max attempts: {}" , DagManagerUtils . getFullyQualifiedJobName ( node ) , jobStatus . getCurrentAttempts ( ) , jobStatus . getMaxAttempts ( ) ) ; submitJob ( node ) ; } } for ( Map . Entry < String , Set < DagNode < JobExecutionPlan > > > entry : nextSubmitted . entrySet ( ) ) { String dagId = entry . getKey ( ) ; Set < DagNode < JobExecutionPlan > > dagNodes = entry . getValue ( ) ; for ( DagNode dagNode : dagNodes ) { addJobState ( dagId , dagNode ) ; } } for ( DagNode < JobExecutionPlan > dagNode : nodesToCleanUp ) { String dagId = DagManagerUtils . generateDagId ( this . jobToDag . get ( dagNode ) ) ; deleteJobState ( dagId , dagNode ) ; } }
Proceed the execution of each dag node based on job status.
Proceed the execution of each dag node based on job status .