StepUtils { public static Set<Step> getAllDependentSteps(Step rootStep, Set<Step> steps) { Set<Step> dependencies = Sets.newHashSet(); Set<Step> immediateDependents = getImmediateDependentSteps(rootStep, steps); for (Step immediateDependent : immediateDependents) { dependencies.add(immediateDependent); dependencies.addAll(getAllDependentSteps(immediateDependent, steps)); } return dependencies; }  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 testGetAllDependentSteps() { Step step1 = new BatchStep("step1"); Step step2 = new BatchStep("step2"); Step step3 = new BatchStep("step3"); Step step4 = new BatchStep("step4"); Step step5 = new BatchStep("step5"); Step step6 = new BatchStep("step6"); step1.configure(ConfigFactory.empty().withValue("dependencies", ConfigValueFactory.fromIterable(Sets.newHashSet()))); step2.configure(ConfigFactory.empty().withValue("dependencies", ConfigValueFactory.fromIterable(Sets.newHashSet("step1")))); step3.configure(ConfigFactory.empty().withValue("dependencies", ConfigValueFactory.fromIterable(Sets.newHashSet("step2")))); step4.configure(ConfigFactory.empty().withValue("dependencies", ConfigValueFactory.fromIterable(Sets.newHashSet("step3")))); step5.configure(ConfigFactory.empty().withValue("dependencies", ConfigValueFactory.fromIterable(Sets.newHashSet()))); step6.configure(ConfigFactory.empty().withValue("dependencies", ConfigValueFactory.fromIterable(Sets.newHashSet("step5")))); Set<Step> steps = Sets.newHashSet(step1, step2, step3, step4, step5, step6); Set<Step> step1AllDependents = StepUtils.getAllDependentSteps(step1, steps); Set<Step> step2AllDependents = StepUtils.getAllDependentSteps(step2, steps); Set<Step> step3AllDependents = StepUtils.getAllDependentSteps(step3, steps); Set<Step> step4AllDependents = StepUtils.getAllDependentSteps(step4, steps); Set<Step> step5AllDependents = StepUtils.getAllDependentSteps(step5, steps); Set<Step> step6AllDependents = StepUtils.getAllDependentSteps(step6, steps); assertEquals(step1AllDependents.size(), 3); assertEquals(step2AllDependents.size(), 2); assertEquals(step3AllDependents.size(), 1); assertEquals(step4AllDependents.size(), 0); assertEquals(step5AllDependents.size(), 1); assertEquals(step6AllDependents.size(), 0); assertEquals(step1AllDependents, Sets.newHashSet(step2, step3, step4)); assertEquals(step2AllDependents, Sets.newHashSet(step3, step4)); assertEquals(step3AllDependents, Sets.newHashSet(step4)); assertEquals(step4AllDependents, Sets.newHashSet()); assertEquals(step5AllDependents, Sets.newHashSet(step6)); assertEquals(step6AllDependents, Sets.newHashSet()); }