@Test(expected = IllegalArgumentException.class) public void testMultipleOutputsToSameSink() throws Exception { flow = Flow.create(getClass().getSimpleName()); input = flow.createInput(new MockStreamDataSource<>()); Dataset<Object> mapped = MapElements.of(input).using(e -> e).output(); Dataset<Pair<Object, Long>> reduced = ReduceByKey .of(mapped) .keyBy(e -> e).reduceBy(values -> 1L) .windowBy(Time.of(Duration.ofSeconds(1))) .output(); Dataset<Pair<Object, Long>> output = Join.of(mapped, reduced) .by(e -> e, Pair::getFirst) .using((Object l, Pair<Object, Long> r, Collector<Long> c) -> { c.collect(r.getSecond()); }) .windowBy(Time.of(Duration.ofSeconds(1))) .output(); ListDataSink<Pair<Object, Long>> sink = ListDataSink.get(); output.persist(sink); reduced.persist(sink); FlowUnfolder.unfold(flow, Executor.getBasicOps()); }
@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);  }