gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
/**
   * Method to submit a {@link URI} for cancellation requsts to the {@link DagManager}.
   * The {@link DagManager} adds the dag to the {@link BlockingQueue} to be picked up by one of the {@link DagManagerThread}s.
   */
Method to submit a  URI for cancellation requsts to the  DagManager.
The  DagManager adds the dag to the  BlockingQueue to be picked up by one of the  DagManagerThreads.
Method to submit a URI for cancellation requsts to the DagManager . The DagManager adds the dag to the BlockingQueue to be picked up by one of the DagManagerThreads .
synchronized public void stopDag(URI uri) throws IOException {
    String flowGroup = FlowConfigResourceLocalHandler.FlowUriUtils.getFlowGroup(uri);
    String flowName = FlowConfigResourceLocalHandler.FlowUriUtils.getFlowName(uri);
    List<Long> flowExecutionIds = this.jobStatusRetriever.getLatestExecutionIdsForFlow(flowName, flowGroup, 10);
    log.info("Found {} flows to cancel.", flowExecutionIds.size());
    for (long flowExecutionId : flowExecutionIds) {
      int queueId = (int) (flowExecutionId % this.numThreads);
      String dagId = DagManagerUtils.generateDagId(flowGroup, flowName, flowExecutionId);
      if (!this.cancelQueue[queueId].offer(dagId)) {
        throw new IOException("Could not add dag " + dagId + " to cancellation queue.");
      }
    }
  }
synchronized public void stopDag ( URI uri ) throws IOException { String flowGroup = FlowConfigResourceLocalHandler . FlowUriUtils . getFlowGroup ( uri ) ; String flowName = FlowConfigResourceLocalHandler . FlowUriUtils . getFlowName ( uri ) ; List < Long > flowExecutionIds = this . jobStatusRetriever . getLatestExecutionIdsForFlow ( flowName , flowGroup , 10 ) ; log . info ( "Found {} flows to cancel." , flowExecutionIds . size ( ) ) ; for ( long flowExecutionId : flowExecutionIds ) { int queueId = ( int ) ( flowExecutionId % this . numThreads ) ; String dagId = DagManagerUtils . generateDagId ( flowGroup , flowName , flowExecutionId ) ; if ( ! this . cancelQueue [ queueId ] . offer ( dagId ) ) { throw new IOException ( "Could not add dag " + dagId + " to cancellation queue." ) ; } } }
Method to submit a {@link URI} for cancellation requsts to the {@link DagManager}.
Method to submit a { @link URI } for cancellation requsts to the { @link DagManager } .