StepUtils { public static Set<Step> mergeLoadedSteps(Set<Step> baseSteps, Step parentStep, Config baseConfig) { if (ConfigUtils.getApplicationConfig(baseConfig).hasPath(Runner.CONFIG_LOADER_PROPERTY)) { Config loaderConfig = ConfigUtils.getApplicationConfig(baseConfig).getConfig(Runner.CONFIG_LOADER_PROPERTY); Config loadedConfig = ComponentFactory.create(ConfigLoader.class, loaderConfig, true).getConfig(); Set<Step> loadedSteps = extractSteps(loadedConfig, true, false); Set<Step> mergeableSteps = StepUtils.getAllDependentSteps(parentStep, loadedSteps); Set<Step> mergedSteps = Sets.newHashSet(baseSteps); Set<Step> stepsToAdd = Sets.newHashSet(); Set<Step> stepsToRemove = Sets.newHashSet(); for (Step mergeableStep : mergeableSteps) { Optional<Step> correspondingBaseStep = StepUtils.getStepForName(mergeableStep.getName(), baseSteps); if (correspondingBaseStep.isPresent() && !correspondingBaseStep.get().getConfig().equals(mergeableStep.getConfig())) { stepsToAdd.add(mergeableStep); stepsToRemove.add(correspondingBaseStep.get()); } else if (!correspondingBaseStep.isPresent()) { stepsToAdd.add(mergeableStep); } } for (Step baseStep : baseSteps) { if (StepUtils.getAllDependentSteps(parentStep, baseSteps).contains(baseStep) && !mergeableSteps.contains(baseStep)) { stepsToRemove.add(baseStep); } } mergedSteps.removeAll(stepsToRemove); mergedSteps.addAll(stepsToAdd); LOG.debug( "Merged loaded steps into stream batch steps: added ({}), removed ({})", StepUtils.stepNamesAsString(stepsToAdd), StepUtils.stepNamesAsString(stepsToRemove)); return mergedSteps; } else { return baseSteps; } }  static boolean allStepsSubmitted(Set<Step> steps); static boolean allStepsFinished(Set<Step> steps); static Set<Step> getDependencies(Step step, Set<Step> steps); static boolean hasStreamingStep(Set<Step> steps); static Set<StreamingStep> getStreamingSteps(Set<Step> steps); static Set<Step> getAllDependentSteps(Step rootStep, Set<Step> steps); static Set<Step> getImmediateDependentSteps(Step step, Set<Step> steps); static Set<Step> getIndependentNonStreamingSteps(Set<Step> steps); static String stepNamesAsString(Set<? extends Step> steps); static void resetSteps(Set<Step> steps); static void resetRepeatingSteps(Set<Step> allSteps); static Set<DataStep> getDataSteps(Set<Step> steps); static Optional<Step> getStepForName(String name, Set<Step> steps); static Set<Step> copySteps(Set<Step> steps); static Map<String, Dataset<Row>> getStepDataFrames(Set<Step> steps); static Set<Step> extractSteps(Config config, boolean configure, boolean notify); static Set<Step> mergeLoadedSteps(Set<Step> baseSteps, Step parentStep, Config baseConfig); static Map<String, StepState> getStepStates(Set<Step> steps); static Set<Step> getStepsMatchingState(Set<Step> steps, StepState matching);  }
@Test public void testMergeLoadedSteps() { Config baseConfig = ConfigUtils.configFromResource("/configuration/step-utils/base.conf").resolve(); Set<Step> baseSteps = StepUtils.extractSteps(baseConfig, true, false); Step parentStep = StepUtils.getStepForName("parent", baseSteps).get(); Set<Step> mergedSteps = StepUtils.mergeLoadedSteps(baseSteps, parentStep, baseConfig); Config expectedConfig = ConfigUtils.configFromResource("/configuration/step-utils/merged.conf").resolve(); Set<Step> expectedSteps = StepUtils.extractSteps(expectedConfig, true, false); assertEquals(expectedSteps.size(), mergedSteps.size()); for (Step expectedStep : expectedSteps) { assertTrue(StepUtils.getStepForName(expectedStep.getName(), mergedSteps).isPresent()); assertEquals(expectedStep.getConfig(), StepUtils.getStepForName(expectedStep.getName(), mergedSteps).get().getConfig()); } }