private void scheduleSpecsFromCatalog() {
    Iterator<URI> specUris = null;
    long startTime = System.currentTimeMillis();

    try {
      specUris = this.flowCatalog.get().getSpecURIs();
    } catch (SpecSerDeException ssde) {
      throw new RuntimeException("Failed to get the iterator of all Spec URIS", ssde);
    }


    try {
      while (specUris.hasNext()) {
        Spec spec = null;
        try {
          spec = this.flowCatalog.get().getSpec(specUris.next());
        } catch (SpecNotFoundException snfe) {
          _log.error(String.format("The URI %s discovered in SpecStore is missing in FlowCatlog"
              + ", suspecting current modification on SpecStore", specUris.next()), snfe);
        }

        //Disable FLOW_RUN_IMMEDIATELY on service startup or leadership change
        if (spec instanceof FlowSpec) {
          Spec modifiedSpec = disableFlowRunImmediatelyOnStart((FlowSpec) spec);
          onAddSpec(modifiedSpec);
        } else {
          onAddSpec(spec);
        }
      }
    } finally {
      this.flowCatalog.get().getMetrics().updateGetSpecTime(startTime);
    }
  }
private void scheduleSpecsFromCatalog ( ) { Iterator < URI > specUris = null ; long startTime = System . currentTimeMillis ( ) ; try { specUris = this . flowCatalog . get ( ) . getSpecURIs ( ) ; } catch ( SpecSerDeException ssde ) { throw new RuntimeException ( "Failed to get the iterator of all Spec URIS" , ssde ) ; } try { while ( specUris . hasNext ( ) ) { Spec spec = null ; try { spec = this . flowCatalog . get ( ) . getSpec ( specUris . next ( ) ) ; } catch ( SpecNotFoundException snfe ) { _log . error ( String . format ( "The URI %s discovered in SpecStore is missing in FlowCatlog" + ", suspecting current modification on SpecStore" , specUris . next ( ) ) , snfe ) ; } if ( spec instanceof FlowSpec ) { Spec modifiedSpec = disableFlowRunImmediatelyOnStart ( ( FlowSpec ) spec ) ; onAddSpec ( modifiedSpec ) ; } else { onAddSpec ( spec ) ; } } } finally { this . flowCatalog . get ( ) . getMetrics ( ) . updateGetSpecTime ( startTime ) ; } }
Load all {@link FlowSpec}s from {@link FlowCatalog} as one of the initialization step,
and make schedulers be aware of that.