@Test(timeout = 5000L) public void testInputMultiConsumption() throws InterruptedException, ExecutionException { final int N = 1000; Dataset<Integer> input = flow.createInput( ListDataSource.unbounded(sequenceInts(0, N))); Dataset<Integer> first = MapElements.of(input) .using(e -> e) .output(); Dataset<Integer> second = MapElements.of(input) .using(e -> e) .output(); Dataset<Integer> union = Union.of(first, second) .output(); Dataset<Pair<Integer, Integer>> sum = ReduceByKey .of(union) .keyBy(e -> 0) .valueBy(e -> e) .reduceBy(Sums.ofInts()) .output(); ListDataSink<Pair<Integer, Integer>> sumOut = ListDataSink.get(); sum.persist(sumOut); executor.submit(flow).get(); DatasetAssert.unorderedEquals( sumOut.getOutputs(), Pair.of(0, 2 * (N - 1) * N / 2)); }
@Override public CompletableFuture<Executor.Result> submit(Flow flow) { return CompletableFuture.supplyAsync(() -> execute(flow), executor); }
LocalExecutor implements Executor { @Override public CompletableFuture<Executor.Result> submit(Flow flow) { return CompletableFuture.supplyAsync(() -> execute(flow), executor); } }
LocalExecutor implements Executor { @Override public CompletableFuture<Executor.Result> submit(Flow flow) { return CompletableFuture.supplyAsync(() -> execute(flow), executor); }  }
LocalExecutor implements Executor { @Override public CompletableFuture<Executor.Result> submit(Flow flow) { return CompletableFuture.supplyAsync(() -> execute(flow), executor); }  LocalExecutor setAllowEarlyEmitting(boolean allowEarlyEmitting); LocalExecutor setWatermarkEmitStrategySupplier(
      java.util.function.Supplier<WatermarkEmitStrategy> supplier); LocalExecutor setTriggeringSchedulerSupplier(
      java.util.function.Supplier<TriggerScheduler> supplier); LocalExecutor allowWindowBasedShuffling(); LocalExecutor setStateContext(StateContext context); @Override CompletableFuture<Executor.Result> submit(Flow flow); @Override void shutdown(); @Override void setAccumulatorProvider(AccumulatorProvider.Factory factory); void abort(); LocalExecutor setDefaultParallelism(int parallel); }
LocalExecutor implements Executor { @Override public CompletableFuture<Executor.Result> submit(Flow flow) { return CompletableFuture.supplyAsync(() -> execute(flow), executor); }  LocalExecutor setAllowEarlyEmitting(boolean allowEarlyEmitting); LocalExecutor setWatermarkEmitStrategySupplier(
      java.util.function.Supplier<WatermarkEmitStrategy> supplier); LocalExecutor setTriggeringSchedulerSupplier(
      java.util.function.Supplier<TriggerScheduler> supplier); LocalExecutor allowWindowBasedShuffling(); LocalExecutor setStateContext(StateContext context); @Override CompletableFuture<Executor.Result> submit(Flow flow); @Override void shutdown(); @Override void setAccumulatorProvider(AccumulatorProvider.Factory factory); void abort(); LocalExecutor setDefaultParallelism(int parallel);  }