container/src/main/java/com/streamsets/datacollector/runner/DetachedStage.java
public<S> DetachedStageRuntime<? extends S> createDetachedStage(
    DetachedStageConfiguration stageConf,
    StageLibraryTask stageLibrary,
    String pipelineId,
    String pipelineTitle,
    String rev,
    Stage.UserContext userContext,
    MetricRegistry metrics,
    ExecutionMode executionMode,
    DeliveryGuarantee deliveryGuarantee,
    RuntimeInfo runtimeInfo,
    EmailSender emailSender,
    Configuration configuration,
    long startTime,
    LineagePublisherDelegator lineagePublisherDelegator,
    Class<S> klass,
    List<Issue> errors
  ) {
    // Firstly validate that the configuration is correct and up to date
    DetachedStageValidator validator = new DetachedStageValidator(stageLibrary, stageConf);
    DetachedStageConfiguration detachedStageConfiguration =  validator.validate();

    // If the stage is not valid, we can't create instance of it
    if(detachedStageConfiguration.getIssues().hasIssues()) {
      errors.addAll(detachedStageConfiguration.getIssues().getIssues());
      return null;
    }

    // Then stageBean that will create new instance and properly propagate all the
    StageBean stageBean = PipelineBeanCreator.get().createStageBean(
      true,
      stageLibrary,
      stageConf.getStageConfiguration(),
      false,
      false,
      false,
      Collections.emptyMap(),
      null,
      errors
    );
    if(!errors.isEmpty()) {
      return null;
    }

    // Stage.Info and Stage.Context
    Stage.Info stageInfo = new Stage.Info() {
      @Override
      public String getName() {
        return stageBean.getDefinition().getName();
      }

      @Override
      public int getVersion() {
        return stageBean.getDefinition().getVersion();
      }

      @Override
      public String getInstanceName() {
        return stageBean.getConfiguration().getInstanceName();
      }

      @Override
      public String getLabel() {
        return stageBean.getConfiguration().getInstanceName();
      }
    };

    StageContext context = new StageContext(
      pipelineId,
      pipelineTitle,
      null, //TODO. Will need to set here if this stage needs to publish lineage events
      rev,
      null, //TODO. Will need to set here if this stage needs to publish lineage events
      Collections.emptyList(),
      userContext,
      stageBean.getDefinition().getType(),
      0,
      false,
      metrics,
      stageBean.getDefinition().getConfigDefinitions(),
      stageBean.getSystemConfigs().stageOnRecordError,
      Collections.emptyList(),
      Collections.emptyMap(),
      stageInfo,
      executionMode,
      deliveryGuarantee,
      runtimeInfo,
      emailSender,
      configuration,
      Collections.emptyMap(),
      startTime,
      lineagePublisherDelegator,
      Collections.emptyMap(),
      false
    );

    return DetachedStageRuntime.create(stageBean, stageInfo, context, klass);
  }
public<S> DetachedStageRuntime<? extends S> createDetachedStage(
    DetachedStageConfiguration stageConf,
    StageLibraryTask stageLibrary,
    String pipelineId,
    String pipelineTitle,
    String rev,
    Stage.UserContext userContext,
    MetricRegistry metrics,
    ExecutionMode executionMode,
    DeliveryGuarantee deliveryGuarantee,
    RuntimeInfo runtimeInfo,
    EmailSender emailSender,
    Configuration configuration,
    long startTime,
    LineagePublisherDelegator lineagePublisherDelegator,
    Class<S> klass,
    List<Issue> errors
  ) {
    // Firstly validate that the configuration is correct and up to date
    DetachedStageValidator validator = new DetachedStageValidator(stageLibrary, stageConf);
    DetachedStageConfiguration detachedStageConfiguration =  validator.validate();

    // If the stage is not valid, we can't create instance of it
    if(detachedStageConfiguration.getIssues().hasIssues()) {
      errors.addAll(detachedStageConfiguration.getIssues().getIssues());
      return null;
    }

    // Then stageBean that will create new instance and properly propagate all the
    StageBean stageBean = PipelineBeanCreator.get().createStageBean(
      true,
      stageLibrary,
      stageConf.getStageConfiguration(),
      false,
      false,
      false,
      Collections.emptyMap(),
      null,
      errors
    );
    if(!errors.isEmpty()) {
      return null;
    }

    // Stage.Info and Stage.Context
    Stage.Info stageInfo = new Stage.Info() {
      @Override
      public String getName() {
        return stageBean.getDefinition().getName();
      }

      @Override
      public int getVersion() {
        return stageBean.getDefinition().getVersion();
      }

      @Override
      public String getInstanceName() {
        return stageBean.getConfiguration().getInstanceName();
      }

      @Override
      public String getLabel() {
        return stageBean.getConfiguration().getInstanceName();
      }
    };

    StageContext context = new StageContext(
      pipelineId,
      pipelineTitle,
      null, //TODO. Will need to set here if this stage needs to publish lineage events
      rev,
      null, //TODO. Will need to set here if this stage needs to publish lineage events
      Collections.emptyList(),
      userContext,
      stageBean.getDefinition().getType(),
      0,
      false,
      metrics,
      stageBean.getDefinition().getConfigDefinitions(),
      stageBean.getSystemConfigs().stageOnRecordError,
      Collections.emptyList(),
      Collections.emptyMap(),
      stageInfo,
      executionMode,
      deliveryGuarantee,
      runtimeInfo,
      emailSender,
      configuration,
      Collections.emptyMap(),
      startTime,
      lineagePublisherDelegator,
      Collections.emptyMap(),
      false
    );

    return DetachedStageRuntime.create(stageBean, stageInfo, context, klass);
  }
public < S > DetachedStageRuntime < ? extends S > createDetachedStage ( DetachedStageConfiguration stageConf , StageLibraryTask stageLibrary , String pipelineId , String pipelineTitle , String rev , Stage . UserContext userContext , MetricRegistry metrics , ExecutionMode executionMode , DeliveryGuarantee deliveryGuarantee , RuntimeInfo runtimeInfo , EmailSender emailSender , Configuration configuration , long startTime , LineagePublisherDelegator lineagePublisherDelegator , Class < S > klass , List < Issue > errors ) { // Firstly validate that the configuration is correct and up to date DetachedStageValidator validator = new DetachedStageValidator ( stageLibrary , stageConf ) ; DetachedStageConfiguration detachedStageConfiguration = validator . validate ( ) ; // If the stage is not valid, we can't create instance of it if ( detachedStageConfiguration . getIssues ( ) . hasIssues ( ) ) { errors . addAll ( detachedStageConfiguration . getIssues ( ) . getIssues ( ) ) ; return null ; } // Then stageBean that will create new instance and properly propagate all the StageBean stageBean = PipelineBeanCreator . get ( ) . createStageBean ( true , stageLibrary , stageConf . getStageConfiguration ( ) , false , false , false , Collections . emptyMap ( ) , null , errors ) ; if ( ! errors . isEmpty ( ) ) { return null ; } // Stage.Info and Stage.Context Stage . Info stageInfo = new Stage . Info ( ) { @ Override public String getName ( ) { return stageBean . getDefinition ( ) . getName ( ) ; } @ Override public int getVersion ( ) { return stageBean . getDefinition ( ) . getVersion ( ) ; } @ Override public String getInstanceName ( ) { return stageBean . getConfiguration ( ) . getInstanceName ( ) ; } @ Override public String getLabel ( ) { return stageBean . getConfiguration ( ) . getInstanceName ( ) ; } } ; StageContext context = new StageContext ( pipelineId , pipelineTitle , null , //TODO. Will need to set here if this stage needs to publish lineage events rev , null , //TODO. Will need to set here if this stage needs to publish lineage events Collections . emptyList ( ) , userContext , stageBean . getDefinition ( ) . getType ( ) , 0 , false , metrics , stageBean . getDefinition ( ) . getConfigDefinitions ( ) , stageBean . getSystemConfigs ( ) . stageOnRecordError , Collections . emptyList ( ) , Collections . emptyMap ( ) , stageInfo , executionMode , deliveryGuarantee , runtimeInfo , emailSender , configuration , Collections . emptyMap ( ) , startTime , lineagePublisherDelegator , Collections . emptyMap ( ) , false ) ; return DetachedStageRuntime . create ( stageBean , stageInfo , context , klass ) ; }
Create a new instance of a stage that does not directly live in the pipeline canvas.
Create a new instance of a stage that does not directly live in the pipeline canvas .