SlidingWindowP extends AbstractProcessor { private static LongStream range(long start, long end, long step) { return start > end ? LongStream.empty() : LongStream.iterate(start, n -> n + step).limit(1 + (end - start) / step); } @SuppressWarnings("unchecked")  SlidingWindowP(
            @Nonnull List<? extends Function<?, ? extends K>> keyFns,
            @Nonnull List<? extends ToLongFunction<?>> frameTimestampFns,
            @Nonnull SlidingWindowPolicy winPolicy,
            long earlyResultsPeriod,
            @Nonnull AggregateOperation<A, ? extends R> aggrOp,
            @Nonnull KeyedWindowResultFunction<? super K, ? super R, ? extends OUT> mapToOutputFn,
            boolean isLastStage
    ); @Override boolean tryProcess(); @Override boolean tryProcessWatermark(@Nonnull Watermark wm); @Override boolean complete(); @Override boolean saveToSnapshot(); @Override boolean finishSnapshotRestore();  }
@Test public void when_receiveRandomTimestamps_then_emitAscending() { final List<Long> timestampsToAdd = LongStream.range(0, 100).boxed().collect(toList()); shuffle(timestampsToAdd); ArrayList<Object> inbox = new ArrayList<>(); for (long ts : timestampsToAdd) { inbox.add(event(ts, 1)); } for (long i = 0; i <= 105; i++) { inbox.add(wm(i)); } List<Object> expectedOutbox = new ArrayList<>(); expectedOutbox.addAll(Arrays.asList( outboxFrame(0, 1), wm(0), outboxFrame(1, 2), wm(1), outboxFrame(2, 3), wm(2), outboxFrame(3, 4), wm(3) )); for (long ts = 4; ts < 100; ts++) { expectedOutbox.add(outboxFrame(ts, 4)); expectedOutbox.add(wm(ts)); } expectedOutbox.addAll(Arrays.asList( outboxFrame(100, 3), wm(100), outboxFrame(101, 2), wm(101), outboxFrame(102, 1), wm(102), wm(103), wm(104), wm(105) )); verifyProcessor(supplier) .disableCompleteCall() .disableLogging() .input(inbox) .expectOutput(expectedOutbox); }