private static void traverseHelper(OperatorSpec startOpSpec, Consumer<OperatorSpec> visitor,
      Function<OperatorSpec, Iterable<OperatorSpec>> getNextOpSpecs, Set<OperatorSpec> visitedOpSpecs) {
    visitor.accept(startOpSpec);
    for (OperatorSpec nextOpSpec : getNextOpSpecs.apply(startOpSpec)) {
      // Make sure we do not end up endlessly traversing cycles.
      if (visitedOpSpecs.add(nextOpSpec)) {
        traverseHelper(nextOpSpec, visitor, getNextOpSpecs, visitedOpSpecs);
      }
    }
  }
Traverses {@link OperatorSpec}s starting from {@code startOpSpec}, invoking {@code visitor} with every encountered
{@link OperatorSpec}, and using {@code getNextOpSpecs} to determine the set of {@link OperatorSpec}s to visit next.
samza-core/src/main/java/org/apache/samza/execution/OperatorSpecGraphAnalyzer.java