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);
    }
  }
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.
Internal method that will upgrade only Stage configuration - not the associated services - and only if needed .