@Test @SuppressWarnings("unchecked") public void testUnfoldBasic() { DAG<Operator<?, ?>> unfolded = FlowUnfolder.unfold(flow, Executor.getBasicOps()); assertEquals(7, unfolded.size()); assertEquals(1, unfolded.getRoots().size()); Node<Operator<?, ?>> root = unfolded.getRoots().stream().findFirst().get(); assertEquals(InputOperator.class, root.get().getClass()); assertEquals(1, root.getChildren().size()); Node<Operator<?, ?>> map = root.getChildren().get(0); assertEquals(FlatMap.class, map.get().getClass()); assertEquals(2, map.getChildren().size()); java.util.Map<Class<? extends Operator>, Node<Operator<?, ?>>> childrenMap; childrenMap = toClassMap((List) map.getChildren()); assertTrue(childrenMap.containsKey(FlatMap.class)); assertTrue(childrenMap.containsKey(ReduceStateByKey.class)); Node<Operator<?, ?>> firstFlatMap = childrenMap.get(FlatMap.class); Node<Operator<?, ?>> firstReduceStateByKey = childrenMap.get( ReduceStateByKey.class); Node<Operator<?, ?>> union = getOnlyAndValidate( firstFlatMap.getChildren(), Union.class); Node<Operator<?, ?>> secondReduceStateByKey = getOnlyAndValidate( union.getChildren(), ReduceStateByKey.class); assertNotNull(secondReduceStateByKey.get().output().getOutputSink()); assertEquals(StdoutSink.class, secondReduceStateByKey.get().output().getOutputSink().getClass()); Node<Operator<?, ?>> secondFlatMap = getOnlyAndValidate( firstReduceStateByKey.getChildren(), FlatMap.class); assertTrue(union == getOnlyAndValidate(secondFlatMap.getChildren(), Union.class)); }
@SuppressWarnings("unchecked") public static DAG<Operator<?, ?>> unfold(Flow flow, Set<Class<? extends Operator<?, ?>>> operatorClasses) throws IllegalArgumentException { return unfold(flow, op -> operatorClasses.contains(op.getClass())); }
FlowUnfolder { @SuppressWarnings("unchecked") public static DAG<Operator<?, ?>> unfold(Flow flow, Set<Class<? extends Operator<?, ?>>> operatorClasses) throws IllegalArgumentException { return unfold(flow, op -> operatorClasses.contains(op.getClass())); } }
FlowUnfolder { @SuppressWarnings("unchecked") public static DAG<Operator<?, ?>> unfold(Flow flow, Set<Class<? extends Operator<?, ?>>> operatorClasses) throws IllegalArgumentException { return unfold(flow, op -> operatorClasses.contains(op.getClass())); }  }
FlowUnfolder { @SuppressWarnings("unchecked") public static DAG<Operator<?, ?>> unfold(Flow flow, Set<Class<? extends Operator<?, ?>>> operatorClasses) throws IllegalArgumentException { return unfold(flow, op -> operatorClasses.contains(op.getClass())); }  @SuppressWarnings("unchecked") static DAG<Operator<?, ?>> unfold(Flow flow,
      Set<Class<? extends Operator<?, ?>>> operatorClasses); static DAG<Operator<?, ?>> unfold(
      Flow flow, UnaryPredicate<Operator<?, ?>> wantTranslate); }
FlowUnfolder { @SuppressWarnings("unchecked") public static DAG<Operator<?, ?>> unfold(Flow flow, Set<Class<? extends Operator<?, ?>>> operatorClasses) throws IllegalArgumentException { return unfold(flow, op -> operatorClasses.contains(op.getClass())); }  @SuppressWarnings("unchecked") static DAG<Operator<?, ?>> unfold(Flow flow,
      Set<Class<? extends Operator<?, ?>>> operatorClasses); static DAG<Operator<?, ?>> unfold(
      Flow flow, UnaryPredicate<Operator<?, ?>> wantTranslate);  }