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;
  }
Upgrade whole Stage configuration, including all services if needed.

This method is idempotent.