container/src/main/java/com/streamsets/datacollector/configupgrade/PipelineConfigurationUpgrader.java
static StageConfiguration upgradeIfNeeded(StageLibraryTask library, StageDefinition def, StageConfiguration conf, List<Issue> issues) {
    IssueCreator issueCreator = IssueCreator.getStage(conf.getInstanceName());
    int fromVersion = conf.getStageVersion();
    int toVersion = def.getVersion();
    try {

      // Firstly upgrade stage itself (register any new services)
      upgradeStageIfNeeded(def, conf, issueCreator, issues);

      // And then upgrade all it's services
      conf.getServices().forEach(serviceConf -> upgradeServicesIfNeeded(
          library,
          conf,
          serviceConf,
          issueCreator.forService(serviceConf.getService().getName()),
          issues
      ));
    } catch (Exception ex) {
      LOG.error("Unknown exception during upgrade: " + ex, ex);
      issues.add(issueCreator.create(
          ContainerError.CONTAINER_0900,
          fromVersion,
          toVersion,
          ex.toString()
      ));
    }

    return conf;
  }
static StageConfiguration upgradeIfNeeded(StageLibraryTask library, StageDefinition def, StageConfiguration conf, List<Issue> issues) {
    IssueCreator issueCreator = IssueCreator.getStage(conf.getInstanceName());
    int fromVersion = conf.getStageVersion();
    int toVersion = def.getVersion();
    try {

      // Firstly upgrade stage itself (register any new services)
      upgradeStageIfNeeded(def, conf, issueCreator, issues);

      // And then upgrade all it's services
      conf.getServices().forEach(serviceConf -> upgradeServicesIfNeeded(
          library,
          conf,
          serviceConf,
          issueCreator.forService(serviceConf.getService().getName()),
          issues
      ));
    } catch (Exception ex) {
      LOG.error("Unknown exception during upgrade: " + ex, ex);
      issues.add(issueCreator.create(
          ContainerError.CONTAINER_0900,
          fromVersion,
          toVersion,
          ex.toString()
      ));
    }

    return conf;
  }
static StageConfiguration upgradeIfNeeded ( StageLibraryTask library , StageDefinition def , StageConfiguration conf , List < Issue > issues ) { IssueCreator issueCreator = IssueCreator . getStage ( conf . getInstanceName ( ) ) ; int fromVersion = conf . getStageVersion ( ) ; int toVersion = def . getVersion ( ) ; try { // Firstly upgrade stage itself (register any new services) upgradeStageIfNeeded ( def , conf , issueCreator , issues ) ; // And then upgrade all it's services conf . getServices ( ) . forEach ( serviceConf -> upgradeServicesIfNeeded ( library , conf , serviceConf , issueCreator . forService ( serviceConf . getService ( ) . getName ( ) ) , issues ) ) ; } catch ( Exception ex ) { LOG . error ( "Unknown exception during upgrade: " + ex , ex ) ; issues . add ( issueCreator . create ( ContainerError . CONTAINER_0900 , fromVersion , toVersion , ex . toString ( ) ) ) ; } return conf ; }
Upgrade whole Stage configuration, including all services if needed.

This method is idempotent.
Upgrade whole Stage configuration including all services if needed .