@Test public void simpleFlatMapTest() throws InterruptedException, ExecutionException { Dataset<Integer> ints = flow.createInput( ListDataSource.unbounded( Arrays.asList(0, 1, 2, 3), Arrays.asList(4, 5, 6))); Dataset<Integer> output = FlatMap.of(ints) .using((Integer e, Collector<Integer> c) -> { for (int i = 0; i < e; i++) { c.collect(e); } }) .output(); ListDataSink<Integer> outputSink = ListDataSink.get(); output.persist(outputSink); executor.submit(flow).get(); List<Integer> outputs = outputSink.getOutputs(); DatasetAssert.unorderedEquals( outputs, 1, 2, 2, 3, 3, 3, 4, 4, 4, 4, 5, 5, 5, 5, 5, 6, 6, 6, 6, 6, 6); }
@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);  }