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;
  }
Internal method that will upgrade service configuration if needed.

This method is idempotent.