gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
/**
     * Perform clean up. Remove a dag from the dagstore if the dag is complete and update internal state.
     */
Perform clean up. Remove a dag from the dagstore if the dag is complete and update internal state.
Perform clean up . Remove a dag from the dagstore if the dag is complete and update internal state .
private void cleanUp() {
      List<String> dagIdstoClean = new ArrayList<>();
      for (String dagId : this.failedDagIdsFinishRunning) {
        LinkedList<DagNode<JobExecutionPlan>> dagNodeList = this.dagToJobs.get(dagId);
        while (!dagNodeList.isEmpty()) {
          DagNode<JobExecutionPlan> dagNode = dagNodeList.poll();
          deleteJobState(dagId, dagNode);
        }
        log.info("Dag {} has finished with status FAILED; Cleaning up dag from the state store.", dagId);
        dagIdstoClean.add(dagId);
      }
      for (String dagId : this.dags.keySet()) {
        if (!hasRunningJobs(dagId) && !this.failedDagIdsFinishRunning.contains(dagId)) {
          String status = "COMPLETE";
          if (this.failedDagIdsFinishAllPossible.contains(dagId)) {
            status = "FAILED";
            this.failedDagIdsFinishAllPossible.remove(dagId);
          }
          log.info("Dag {} has finished with status {}; Cleaning up dag from the state store.", dagId, status);
          dagIdstoClean.add(dagId);
        }
      }
      for (String dagId: dagIdstoClean) {
        cleanUpDag(dagId);
      }
    }
private void cleanUp ( ) { List < String > dagIdstoClean = new ArrayList < > ( ) ; for ( String dagId : this . failedDagIdsFinishRunning ) { LinkedList < DagNode < JobExecutionPlan > > dagNodeList = this . dagToJobs . get ( dagId ) ; while ( ! dagNodeList . isEmpty ( ) ) { DagNode < JobExecutionPlan > dagNode = dagNodeList . poll ( ) ; deleteJobState ( dagId , dagNode ) ; } log . info ( "Dag {} has finished with status FAILED; Cleaning up dag from the state store." , dagId ) ; dagIdstoClean . add ( dagId ) ; } for ( String dagId : this . dags . keySet ( ) ) { if ( ! hasRunningJobs ( dagId ) && ! this . failedDagIdsFinishRunning . contains ( dagId ) ) { String status = "COMPLETE" ; if ( this . failedDagIdsFinishAllPossible . contains ( dagId ) ) { status = "FAILED" ; this . failedDagIdsFinishAllPossible . remove ( dagId ) ; } log . info ( "Dag {} has finished with status {}; Cleaning up dag from the state store." , dagId , status ) ; dagIdstoClean . add ( dagId ) ; } } for ( String dagId : dagIdstoClean ) { cleanUpDag ( dagId ) ; } }
Perform clean up.
Perform clean up .
//Clean up failed dags //Skip monitoring of any other jobs of the failed dag. //Clean up completed dags