jms-lib/src/main/java/com/streamsets/pipeline/stage/origin/jms/JmsSourceUpgrader.java
private static void upgradeV5ToV6(List<Config> configs, Context context) {
    List<Config> dataFormatConfigs = configs.stream()
      .filter(c -> c.getName().startsWith("dataFormat"))
      .collect(Collectors.toList());

    // Remove those configs
    configs.removeAll(dataFormatConfigs);

    // There is an interesting history with compression - at some point (version 2), we explicitly added it, then
    // we have hidden it. So this config might or might not exists, depending on the version in which the pipeline
    // was created. However the service is expecting it and thus we need to ensure that it's there.
    if(dataFormatConfigs.stream().noneMatch(c -> "dataFormatConfig.compression".equals(c.getName()))) {
      dataFormatConfigs.add(new Config("dataFormatConfig.compression", "NONE"));
    }

    // And finally register new service
    context.registerService(DataFormatParserService.class, dataFormatConfigs);
  }
private static void upgradeV5ToV6(List<Config> configs, Context context) {
    List<Config> dataFormatConfigs = configs.stream()
      .filter(c -> c.getName().startsWith("dataFormat"))
      .collect(Collectors.toList());

    // Remove those configs
    configs.removeAll(dataFormatConfigs);

    // There is an interesting history with compression - at some point (version 2), we explicitly added it, then
    // we have hidden it. So this config might or might not exists, depending on the version in which the pipeline
    // was created. However the service is expecting it and thus we need to ensure that it's there.
    if(dataFormatConfigs.stream().noneMatch(c -> "dataFormatConfig.compression".equals(c.getName()))) {
      dataFormatConfigs.add(new Config("dataFormatConfig.compression", "NONE"));
    }

    // And finally register new service
    context.registerService(DataFormatParserService.class, dataFormatConfigs);
  }
Transition to services