container/src/main/java/com/streamsets/datacollector/creation/PipelineBeanCreator.java
public PipelineBean create(
      boolean forExecution,
      StageLibraryTask library,
      PipelineConfiguration pipelineConf,
      InterceptorCreatorContextBuilder interceptorContextBuilder,
      List<Issue> errors,
      Map<String, Object> runtimeParameters
  ) {
    int priorErrors = errors.size();
    PipelineConfigBean pipelineConfigBean = create(pipelineConf, errors, runtimeParameters);
    StageBean errorStageBean = null;
    StageBean statsStageBean = null;
    StageBean origin = null;
    PipelineStageBeans stages = null;
    PipelineStageBeans startEventBeans = null;
    PipelineStageBeans stopEventBeans = null;
    if (pipelineConfigBean != null && pipelineConfigBean.constants != null) {
      Map<String, Object> resolvedConstants = pipelineConfigBean.constants;
      if(interceptorContextBuilder != null) {
        interceptorContextBuilder
          .withExecutionMode(pipelineConfigBean.executionMode)
          .withDeliveryGuarantee(pipelineConfigBean.deliveryGuarantee)
        ;
      }

      // Instantiate usual stages
      if(!pipelineConf.getStages().isEmpty()) {
        origin = createStageBean(
            forExecution,
            library,
            pipelineConf.getStages().get(0),
            true,
            false,
            false,
            resolvedConstants,
            interceptorContextBuilder,
            errors
        );

        stages = createPipelineStageBeans(
            forExecution,
            library,
            pipelineConf.getStages().subList(1, pipelineConf.getStages().size()),
            interceptorContextBuilder,
            resolvedConstants,
            errors
        );
      }

      // It is not mandatory to have a stats aggregating target configured
      StageConfiguration statsStageConf = pipelineConf.getStatsAggregatorStage();
      if (statsStageConf != null) {
        statsStageBean = createStageBean(
            forExecution,
            library,
            statsStageConf,
            true,
            false,
            false,
            resolvedConstants,
            interceptorContextBuilder,
            errors
        );
      }

      // Error stage is mandatory
      StageConfiguration errorStageConf = pipelineConf.getErrorStage();
      if (errorStageConf != null) {
        errorStageBean = createStageBean(
            forExecution,
            library,
            errorStageConf,
            true,
            true,
            false,
            resolvedConstants,
            interceptorContextBuilder,
            errors
        );
      } else if (!(pipelineConfigBean.executionMode.equals(ExecutionMode.BATCH) ||
          pipelineConfigBean.executionMode.equals(ExecutionMode.STREAMING))) {
        errors.add(IssueCreator.getPipeline().create(
            PipelineGroups.BAD_RECORDS.name(),
            "badRecordsHandling",
            CreationError.CREATION_009
        ));
      }

      // Pipeline Lifecycle event handlers
      StageBean startBean = null;
      if(CollectionUtils.isNotEmpty(pipelineConf.getStartEventStages())) {
         startBean = createStageBean(
            forExecution,
            library,
            pipelineConf.getStartEventStages().get(0),
            true,
            false,
            true,
            resolvedConstants,
            interceptorContextBuilder,
            errors
        );
      }
      startEventBeans = new PipelineStageBeans(startBean == null ? Collections.emptyList() : ImmutableList.of(startBean));
      StageBean stopBean = null;
      if(CollectionUtils.isNotEmpty(pipelineConf.getStopEventStages())) {
         stopBean = createStageBean(
            forExecution,
            library,
            pipelineConf.getStopEventStages().get(0),
            true,
            false,
            true,
            resolvedConstants,
            interceptorContextBuilder,
            errors
        );
      }
      stopEventBeans = new PipelineStageBeans(stopBean == null ? Collections.emptyList() : ImmutableList.of(stopBean));

      // Validate Webhook Configs
      if (pipelineConfigBean.webhookConfigs != null && !pipelineConfigBean.webhookConfigs.isEmpty()) {
        int index = 0;
        for (PipelineWebhookConfig webhookConfig: pipelineConfigBean.webhookConfigs) {
          if (StringUtils.isEmpty(webhookConfig.webhookUrl)) {
            Issue issue = IssueCreator.getPipeline().create(
                PipelineGroups.NOTIFICATIONS.name(),
                "webhookUrl",
                CreationError.CREATION_080
            );
            issue.setAdditionalInfo("index", index);
            errors.add(issue);
            break;
          }
          index++;
        }
      }
    }

    // Something went wrong
    if(errors.size() != priorErrors) {
      return  null;
    }

    return new PipelineBean(
        pipelineConfigBean,
        origin,
        stages,
        errorStageBean,
        statsStageBean,
        startEventBeans,
        stopEventBeans
    );
  }
public PipelineBean create(
      boolean forExecution,
      StageLibraryTask library,
      PipelineConfiguration pipelineConf,
      InterceptorCreatorContextBuilder interceptorContextBuilder,
      List<Issue> errors,
      Map<String, Object> runtimeParameters
  ) {
    int priorErrors = errors.size();
    PipelineConfigBean pipelineConfigBean = create(pipelineConf, errors, runtimeParameters);
    StageBean errorStageBean = null;
    StageBean statsStageBean = null;
    StageBean origin = null;
    PipelineStageBeans stages = null;
    PipelineStageBeans startEventBeans = null;
    PipelineStageBeans stopEventBeans = null;
    if (pipelineConfigBean != null && pipelineConfigBean.constants != null) {
      Map<String, Object> resolvedConstants = pipelineConfigBean.constants;
      if(interceptorContextBuilder != null) {
        interceptorContextBuilder
          .withExecutionMode(pipelineConfigBean.executionMode)
          .withDeliveryGuarantee(pipelineConfigBean.deliveryGuarantee)
        ;
      }

      // Instantiate usual stages
      if(!pipelineConf.getStages().isEmpty()) {
        origin = createStageBean(
            forExecution,
            library,
            pipelineConf.getStages().get(0),
            true,
            false,
            false,
            resolvedConstants,
            interceptorContextBuilder,
            errors
        );

        stages = createPipelineStageBeans(
            forExecution,
            library,
            pipelineConf.getStages().subList(1, pipelineConf.getStages().size()),
            interceptorContextBuilder,
            resolvedConstants,
            errors
        );
      }

      // It is not mandatory to have a stats aggregating target configured
      StageConfiguration statsStageConf = pipelineConf.getStatsAggregatorStage();
      if (statsStageConf != null) {
        statsStageBean = createStageBean(
            forExecution,
            library,
            statsStageConf,
            true,
            false,
            false,
            resolvedConstants,
            interceptorContextBuilder,
            errors
        );
      }

      // Error stage is mandatory
      StageConfiguration errorStageConf = pipelineConf.getErrorStage();
      if (errorStageConf != null) {
        errorStageBean = createStageBean(
            forExecution,
            library,
            errorStageConf,
            true,
            true,
            false,
            resolvedConstants,
            interceptorContextBuilder,
            errors
        );
      } else if (!(pipelineConfigBean.executionMode.equals(ExecutionMode.BATCH) ||
          pipelineConfigBean.executionMode.equals(ExecutionMode.STREAMING))) {
        errors.add(IssueCreator.getPipeline().create(
            PipelineGroups.BAD_RECORDS.name(),
            "badRecordsHandling",
            CreationError.CREATION_009
        ));
      }

      // Pipeline Lifecycle event handlers
      StageBean startBean = null;
      if(CollectionUtils.isNotEmpty(pipelineConf.getStartEventStages())) {
         startBean = createStageBean(
            forExecution,
            library,
            pipelineConf.getStartEventStages().get(0),
            true,
            false,
            true,
            resolvedConstants,
            interceptorContextBuilder,
            errors
        );
      }
      startEventBeans = new PipelineStageBeans(startBean == null ? Collections.emptyList() : ImmutableList.of(startBean));
      StageBean stopBean = null;
      if(CollectionUtils.isNotEmpty(pipelineConf.getStopEventStages())) {
         stopBean = createStageBean(
            forExecution,
            library,
            pipelineConf.getStopEventStages().get(0),
            true,
            false,
            true,
            resolvedConstants,
            interceptorContextBuilder,
            errors
        );
      }
      stopEventBeans = new PipelineStageBeans(stopBean == null ? Collections.emptyList() : ImmutableList.of(stopBean));

      // Validate Webhook Configs
      if (pipelineConfigBean.webhookConfigs != null && !pipelineConfigBean.webhookConfigs.isEmpty()) {
        int index = 0;
        for (PipelineWebhookConfig webhookConfig: pipelineConfigBean.webhookConfigs) {
          if (StringUtils.isEmpty(webhookConfig.webhookUrl)) {
            Issue issue = IssueCreator.getPipeline().create(
                PipelineGroups.NOTIFICATIONS.name(),
                "webhookUrl",
                CreationError.CREATION_080
            );
            issue.setAdditionalInfo("index", index);
            errors.add(issue);
            break;
          }
          index++;
        }
      }
    }

    // Something went wrong
    if(errors.size() != priorErrors) {
      return  null;
    }

    return new PipelineBean(
        pipelineConfigBean,
        origin,
        stages,
        errorStageBean,
        statsStageBean,
        startEventBeans,
        stopEventBeans
    );
  }
public PipelineBean create ( boolean forExecution , StageLibraryTask library , PipelineConfiguration pipelineConf , InterceptorCreatorContextBuilder interceptorContextBuilder , List < Issue > errors , Map < String , Object > runtimeParameters ) { int priorErrors = errors . size ( ) ; PipelineConfigBean pipelineConfigBean = create ( pipelineConf , errors , runtimeParameters ) ; StageBean errorStageBean = null ; StageBean statsStageBean = null ; StageBean origin = null ; PipelineStageBeans stages = null ; PipelineStageBeans startEventBeans = null ; PipelineStageBeans stopEventBeans = null ; if ( pipelineConfigBean != null && pipelineConfigBean . constants != null ) { Map < String , Object > resolvedConstants = pipelineConfigBean . constants ; if ( interceptorContextBuilder != null ) { interceptorContextBuilder . withExecutionMode ( pipelineConfigBean . executionMode ) . withDeliveryGuarantee ( pipelineConfigBean . deliveryGuarantee ) ; } // Instantiate usual stages if ( ! pipelineConf . getStages ( ) . isEmpty ( ) ) { origin = createStageBean ( forExecution , library , pipelineConf . getStages ( ) . get ( 0 ) , true , false , false , resolvedConstants , interceptorContextBuilder , errors ) ; stages = createPipelineStageBeans ( forExecution , library , pipelineConf . getStages ( ) . subList ( 1 , pipelineConf . getStages ( ) . size ( ) ) , interceptorContextBuilder , resolvedConstants , errors ) ; } // It is not mandatory to have a stats aggregating target configured StageConfiguration statsStageConf = pipelineConf . getStatsAggregatorStage ( ) ; if ( statsStageConf != null ) { statsStageBean = createStageBean ( forExecution , library , statsStageConf , true , false , false , resolvedConstants , interceptorContextBuilder , errors ) ; } // Error stage is mandatory StageConfiguration errorStageConf = pipelineConf . getErrorStage ( ) ; if ( errorStageConf != null ) { errorStageBean = createStageBean ( forExecution , library , errorStageConf , true , true , false , resolvedConstants , interceptorContextBuilder , errors ) ; } else if ( ! ( pipelineConfigBean . executionMode . equals ( ExecutionMode . BATCH ) || pipelineConfigBean . executionMode . equals ( ExecutionMode . STREAMING ) ) ) { errors . add ( IssueCreator . getPipeline ( ) . create ( PipelineGroups . BAD_RECORDS . name ( ) , "badRecordsHandling" , CreationError . CREATION_009 ) ) ; } // Pipeline Lifecycle event handlers StageBean startBean = null ; if ( CollectionUtils . isNotEmpty ( pipelineConf . getStartEventStages ( ) ) ) { startBean = createStageBean ( forExecution , library , pipelineConf . getStartEventStages ( ) . get ( 0 ) , true , false , true , resolvedConstants , interceptorContextBuilder , errors ) ; } startEventBeans = new PipelineStageBeans ( startBean == null ? Collections . emptyList ( ) : ImmutableList . of ( startBean ) ) ; StageBean stopBean = null ; if ( CollectionUtils . isNotEmpty ( pipelineConf . getStopEventStages ( ) ) ) { stopBean = createStageBean ( forExecution , library , pipelineConf . getStopEventStages ( ) . get ( 0 ) , true , false , true , resolvedConstants , interceptorContextBuilder , errors ) ; } stopEventBeans = new PipelineStageBeans ( stopBean == null ? Collections . emptyList ( ) : ImmutableList . of ( stopBean ) ) ; // Validate Webhook Configs if ( pipelineConfigBean . webhookConfigs != null && ! pipelineConfigBean . webhookConfigs . isEmpty ( ) ) { int index = 0 ; for ( PipelineWebhookConfig webhookConfig : pipelineConfigBean . webhookConfigs ) { if ( StringUtils . isEmpty ( webhookConfig . webhookUrl ) ) { Issue issue = IssueCreator . getPipeline ( ) . create ( PipelineGroups . NOTIFICATIONS . name ( ) , "webhookUrl" , CreationError . CREATION_080 ) ; issue . setAdditionalInfo ( "index" , index ) ; errors . add ( issue ) ; break ; } index ++ ; } } } // Something went wrong if ( errors . size ( ) != priorErrors ) { return null ; } return new PipelineBean ( pipelineConfigBean , origin , stages , errorStageBean , statsStageBean , startEventBeans , stopEventBeans ) ; }
Create PipelineBean which means instantiating all stages for the pipeline.

For multi-threaded pipelines this method will *NOT* create all required instances since the number of required
instances is not known at the time of creation - for that the origin has to be initialized which is not at this
point. Hence this method will create only one instance of the whole pipeline and it's up to the caller to call
createPipelineStageBeans to instantiate remaining source-less pipelines later in the execution.
Create PipelineBean which means instantiating all stages for the pipeline .