@Test public void testReduceByKeyWithSortStateAndCustomWindowing() throws InterruptedException, ExecutionException { Dataset<Integer> ints = flow.createInput( ListDataSource.unbounded( reversed(sequenceInts(0, 100)), reversed(sequenceInts(100, 1100)))); SizedCountWindowing<Integer> windowing = new SizedCountWindowing<>(i -> (i % 10) + 1); Dataset<Pair<Integer, Integer>> output = ReduceStateByKey.of(ints) .keyBy(i -> i % 10) .valueBy(e -> e) .stateFactory(SortState::new) .mergeStatesBy(SortState::combine) .windowBy(windowing) .output(); ListDataSink<Triple<SizedCountWindow, Integer, Integer>> sink = ListDataSink.get(); FlatMap.of(output) .using((UnaryFunctor<Pair<Integer, Integer>, Triple<SizedCountWindow, Integer, Integer>>) (elem, context) -> context.collect(Triple.of((SizedCountWindow) context.getWindow(), elem.getFirst(), elem.getSecond()))) .output() .persist(sink); executor.submit(flow).get(); List<Triple<SizedCountWindow, Integer, Integer>> outputs = sink.getOutputs(); assertEquals(4 * 550, outputs.size()); checkKeyAlignedSortedList(outputs); }
@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);  }