container/src/main/java/com/streamsets/datacollector/configupgrade/PipelineConfigurationUpgrader.java
static private void upgradeStageIfNeeded(
      StageDefinition def,
      StageConfiguration conf,
      IssueCreator issueCreator,
      List<Issue> issues
  ) {
    int fromVersion = conf.getStageVersion();
    int toVersion = def.getVersion();

    // In case we don't need an upgrade
    if(!needsUpgrade(toVersion, fromVersion, IssueCreator.getStage(conf.getInstanceName()), issues)) {
      return;
    }

    ClassLoader cl = Thread.currentThread().getContextClassLoader();
    try {
      Thread.currentThread().setContextClassLoader(def.getStageClassLoader());
      LOG.warn("Upgrading stage instance '{}' from version '{}' to version '{}'", conf.getInstanceName(), fromVersion, toVersion);

      UpgradeContext upgradeContext = new UpgradeContext(
          def.getLibrary(),
          def.getName(),
          conf.getInstanceName(),
          fromVersion,
          toVersion
      );
      List<Config> configs = def.getUpgrader().upgrade(conf.getConfiguration(), upgradeContext);

      conf.setStageVersion(def.getVersion());
      conf.setConfig(configs);

      // Propagate newly registered services to the StageConfiguration
      if(!upgradeContext.registeredServices.isEmpty()) {
        List<ServiceConfiguration> services = new ArrayList<>();
        services.addAll(conf.getServices());

        // Version -1 is special to note that this version has been created by stage and not by the service itself
        upgradeContext.registeredServices
            .forEach((s, c) -> services.add(new ServiceConfiguration(s, -1, c)));

        conf.setServices(services);

      }
    } catch (StageException ex) {
      issues.add(issueCreator.create(ex.getErrorCode(), ex.getParams()));
    } finally {
      Thread.currentThread().setContextClassLoader(cl);
    }
  }
static private void upgradeStageIfNeeded(
      StageDefinition def,
      StageConfiguration conf,
      IssueCreator issueCreator,
      List<Issue> issues
  ) {
    int fromVersion = conf.getStageVersion();
    int toVersion = def.getVersion();

    // In case we don't need an upgrade
    if(!needsUpgrade(toVersion, fromVersion, IssueCreator.getStage(conf.getInstanceName()), issues)) {
      return;
    }

    ClassLoader cl = Thread.currentThread().getContextClassLoader();
    try {
      Thread.currentThread().setContextClassLoader(def.getStageClassLoader());
      LOG.warn("Upgrading stage instance '{}' from version '{}' to version '{}'", conf.getInstanceName(), fromVersion, toVersion);

      UpgradeContext upgradeContext = new UpgradeContext(
          def.getLibrary(),
          def.getName(),
          conf.getInstanceName(),
          fromVersion,
          toVersion
      );
      List<Config> configs = def.getUpgrader().upgrade(conf.getConfiguration(), upgradeContext);

      conf.setStageVersion(def.getVersion());
      conf.setConfig(configs);

      // Propagate newly registered services to the StageConfiguration
      if(!upgradeContext.registeredServices.isEmpty()) {
        List<ServiceConfiguration> services = new ArrayList<>();
        services.addAll(conf.getServices());

        // Version -1 is special to note that this version has been created by stage and not by the service itself
        upgradeContext.registeredServices
            .forEach((s, c) -> services.add(new ServiceConfiguration(s, -1, c)));

        conf.setServices(services);

      }
    } catch (StageException ex) {
      issues.add(issueCreator.create(ex.getErrorCode(), ex.getParams()));
    } finally {
      Thread.currentThread().setContextClassLoader(cl);
    }
  }
Internal method that will upgrade only Stage configuration - not the associated services - and only if needed.

This method is idempotent.