@Nonnull
    public static <T> AggregateOperation1<T, PriorityQueue<T>, List<T>> topN(
            int n, @Nonnull ComparatorEx<? super T> comparator
    ) {
        checkSerializable(comparator, "comparator");
        ComparatorEx<? super T> comparatorReversed = comparator.reversed();
        BiConsumerEx<PriorityQueue<T>, T> accumulateFn = (queue, item) -> {
            if (queue.size() == n) {
                if (comparator.compare(item, queue.peek()) <= 0) {
                    // the new item is smaller or equal to the smallest in queue
                    return;
                }
                queue.poll();
            }
            queue.offer(item);
        };
        return AggregateOperation
                .withCreate(() -> new PriorityQueue<T>(n, comparator))
                .andAccumulate(accumulateFn)
                .andCombine((left, right) -> {
                    for (T item : right) {
                        accumulateFn.accept(left, item);
                    }
                })
                .andExportFinish(queue -> {
                    ArrayList<T> res = new ArrayList<>(queue);
                    res.sort(comparatorReversed);
                    return res;
                });
    }
@Test
    public void when_topN() {
        AggregateOperationsTest.validateOpWithoutDeduct(topN(2, naturalOrder()),
                ArrayList::new,
                Arrays.asList(7, 1, 5, 3),
                Arrays.asList(8, 6, 2, 4),
                Arrays.asList(5, 7),
                Arrays.asList(7, 8),
                Arrays.asList(8, 7)
        );
    }