container/src/main/java/com/streamsets/datacollector/configupgrade/PipelineConfigurationUpgrader.java
private static ServiceConfiguration upgradeServicesIfNeeded(
      StageLibraryTask library,
      StageConfiguration stageConf,
      ServiceConfiguration conf,
      IssueCreator issueCreator,
      List<Issue> issues
  ) {
    ServiceDefinition def = library.getServiceDefinition(conf.getService(), false);
    if (def == null) {
      issues.add(issueCreator.create(ContainerError.CONTAINER_0903, conf.getService().getName()));
    }

    int fromVersion = conf.getServiceVersion();
    int toVersion = def.getVersion();

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

    ClassLoader cl = Thread.currentThread().getContextClassLoader();
    try {
      LOG.warn("Upgrading service instance from version '{}' to version '{}'", conf.getServiceVersion(), def.getVersion());

      UpgradeContext upgradeContext = new UpgradeContext(
          "",
          def.getName(),
          stageConf.getInstanceName(),
          fromVersion,
          toVersion
      );

      List<Config> configs = def.getUpgrader().upgrade(conf.getConfiguration(), upgradeContext);

      if(!upgradeContext.registeredServices.isEmpty()) {
        throw new StageException(ContainerError.CONTAINER_0904);
      }

      conf.setServiceVersion(toVersion);
      conf.setConfig(configs);
    } catch (StageException ex) {
      issues.add(issueCreator.create(ex.getErrorCode(), ex.getParams()));
    } finally {
      Thread.currentThread().setContextClassLoader(cl);
    }

    return conf;
  }
private static ServiceConfiguration upgradeServicesIfNeeded(
      StageLibraryTask library,
      StageConfiguration stageConf,
      ServiceConfiguration conf,
      IssueCreator issueCreator,
      List<Issue> issues
  ) {
    ServiceDefinition def = library.getServiceDefinition(conf.getService(), false);
    if (def == null) {
      issues.add(issueCreator.create(ContainerError.CONTAINER_0903, conf.getService().getName()));
    }

    int fromVersion = conf.getServiceVersion();
    int toVersion = def.getVersion();

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

    ClassLoader cl = Thread.currentThread().getContextClassLoader();
    try {
      LOG.warn("Upgrading service instance from version '{}' to version '{}'", conf.getServiceVersion(), def.getVersion());

      UpgradeContext upgradeContext = new UpgradeContext(
          "",
          def.getName(),
          stageConf.getInstanceName(),
          fromVersion,
          toVersion
      );

      List<Config> configs = def.getUpgrader().upgrade(conf.getConfiguration(), upgradeContext);

      if(!upgradeContext.registeredServices.isEmpty()) {
        throw new StageException(ContainerError.CONTAINER_0904);
      }

      conf.setServiceVersion(toVersion);
      conf.setConfig(configs);
    } catch (StageException ex) {
      issues.add(issueCreator.create(ex.getErrorCode(), ex.getParams()));
    } finally {
      Thread.currentThread().setContextClassLoader(cl);
    }

    return conf;
  }
private static ServiceConfiguration upgradeServicesIfNeeded ( StageLibraryTask library , StageConfiguration stageConf , ServiceConfiguration conf , IssueCreator issueCreator , List < Issue > issues ) { ServiceDefinition def = library . getServiceDefinition ( conf . getService ( ) , false ) ; if ( def == null ) { issues . add ( issueCreator . create ( ContainerError . CONTAINER_0903 , conf . getService ( ) . getName ( ) ) ) ; } int fromVersion = conf . getServiceVersion ( ) ; int toVersion = def . getVersion ( ) ; // In case we don't need an upgrade if ( ! needsUpgrade ( toVersion , fromVersion , issueCreator , issues ) ) { return conf ; } ClassLoader cl = Thread . currentThread ( ) . getContextClassLoader ( ) ; try { LOG . warn ( "Upgrading service instance from version '{}' to version '{}'" , conf . getServiceVersion ( ) , def . getVersion ( ) ) ; UpgradeContext upgradeContext = new UpgradeContext ( "" , def . getName ( ) , stageConf . getInstanceName ( ) , fromVersion , toVersion ) ; List < Config > configs = def . getUpgrader ( ) . upgrade ( conf . getConfiguration ( ) , upgradeContext ) ; if ( ! upgradeContext . registeredServices . isEmpty ( ) ) { throw new StageException ( ContainerError . CONTAINER_0904 ) ; } conf . setServiceVersion ( toVersion ) ; conf . setConfig ( configs ) ; } catch ( StageException ex ) { issues . add ( issueCreator . create ( ex . getErrorCode ( ) , ex . getParams ( ) ) ) ; } finally { Thread . currentThread ( ) . setContextClassLoader ( cl ) ; } return conf ; }
Internal method that will upgrade service configuration if needed.

This method is idempotent.
Internal method that will upgrade service configuration if needed .