Java Stream Gatherers (JEP 485): windowFixed, windowSliding, fold, scan, mapConcurrent and Custom Gatherers

Every intermediate operation on a Stream — filter, map, sorted, distinct — does one thing to one element at a time and hands the result to the next stage. That covers most pipelines. It does not cover “group these into chunks of three”, “give me a running total”, or “call this slow HTTP endpoint for each element, five at a time, without blocking a platform thread per call”. For years the answer was: write a loop, or fake it with a stateful lambda captured in an array, or collect to a List and start a second stream over the result. All three work. None of them are a stream operation anymore.

JEP 485, final since JDK 24 and unchanged in 25, adds a real answer: Stream.gather(Gatherer), a new kind of intermediate operation you can both use off the shelf and write yourself. This post covers all five built-in gatherers with real output, writes two custom ones from scratch, and spends real time on the one surprising behavior a built-in gatherer has when something goes wrong — because that part isn’t in the JEP’s prose and cost a debugging session to find.

Versions used in this post. JDK 25 (Temurin 25.0.4.1+1) — java.util.stream.Gatherers and Gatherer need no --enable-preview flag; JEP 485 finalized in JDK 24 and every example here compiles and runs unchanged on 25. Every code block links to a source file, and every output block is quoted verbatim from a transcript committed alongside it — both live in the javademos repository.

The five built-in gatherers

Stream.gather(Gatherer) sits exactly where filter or map sits: it takes a stream, returns a stream, and can be followed by more intermediate operations or a terminal one. The difference is what a Gatherer is allowed to do that those simpler operations can’t — hold state across elements, emit a different number of outputs than inputs, look at more than one element at a time, and even decide to stop pulling from the source early. java.util.stream.Gatherers ships five ready-made ones covering the transformations people reach for most.

windowFixed(n) chunks the stream into lists of exactly n elements, shrinking only the last window if the source doesn’t divide evenly. windowSliding(n) advances by one each time, so consecutive windows overlap. fold and scan share a signature — a Supplier<R> seed and a BiFunction<R,T,R> — but not a shape: fold emits exactly one final value, scan emits every intermediate one.

List<List<Integer>> windows = Stream.of(1, 2, 3, 4, 5, 6, 7)
        .gather(Gatherers.windowFixed(3))
        .toList();
// [[1, 2, 3], [4, 5, 6], [7]]

List<Integer> runningTotals = Stream.of(1, 2, 3, 4, 5)
        .gather(Gatherers.scan(() -> 0, Integer::sum))
        .toList();
// [1, 3, 6, 10, 15]

Full source: GathererBasics.java. Real, executed output:

windowFixed(3) of 1..7 = [[1, 2, 3], [4, 5, 6], [7]]
CHECK ok   : windowFixed(3) groups into [1,2,3][4,5,6][7], last window is the short remainder
CHECK ok   : windowFixed(0) throws IllegalArgumentException: 'windowSize' must be greater than zero
windowSliding(3) of 1..5 = [[1, 2, 3], [2, 3, 4], [3, 4, 5]]
CHECK ok   : windowSliding(3) of 5 elements produces exactly 3 overlapping windows, each length 3
CHECK ok   : windowSliding(3) on a 2-element stream still emits one short window: [[1, 2]]
CHECK ok   : windowSliding(3) on an empty stream emits nothing
fold building a String: [1][2][3][4]
CHECK ok   : fold("", acc+"[n]") over 1..4 builds "[1][2][3][4]" left to right
scan running total of 1..5 = [1, 3, 6, 10, 15]
CHECK ok   : scan(0, +) over 1..5 emits every prefix sum: [1,3,6,10,15]

windowSliding doesn’t shrink the way windowFixed does. It’s easy to assume the two share a “shrink the leftover” rule. They don’t: on a stream shorter than the window size, windowFixed shrinks its last window, but windowSliding still emits one window holding whatever’s there — windowSliding(3) over a 2-element stream gives [[1, 2]], not an empty result. Only an empty source stream produces zero windows either way. This isn’t documented clearly enough to guess correctly; I checked it by running it.

Two things worth another paragraph before moving on. First, fold and scan have no combiner, which is precisely what lets their result type R differ from the element type T — Stream.reduce can’t safely do that in a way that also supports parallel streams, so it doesn’t offer the option; fold/scan are sequential-only by design and trade parallelism for that flexibility. Second, windowFixed(0) rejects immediately with IllegalArgumentException rather than silently returning nothing, which matters if a window size ever comes from a config value you haven’t validated.

• Reference: the JEP 485 specification for the full built-in list and formal semantics.
• Deeper: chapter 16 of the javademos docs has the exact exception message text and the empty-stream edge case for every built-in gatherer.

mapConcurrent: the one gatherer that’s actually about concurrency

Gatherers.mapConcurrent(maxConcurrency, mapper) is the odd one out — the other four reshape a stream’s elements; this one changes how the mapper runs. Each element gets its own virtual thread, bounded to at most maxConcurrency running at once, with output back in encounter order regardless of which finishes first.

List<Integer> results = Stream.of(1, 2, 3, 4)
        .gather(Gatherers.mapConcurrent(4, n -> {
            sleep(Duration.ofMillis((5 - n) * 40L));   // task 1 finishes LAST
            return n * 10;
        }))
        .toList();
// [10, 20, 30, 40] -- input order, even though completion order was reversed

Source: MapConcurrentDemo.java. This run also caps concurrency at 2 across 8 tasks and instruments the peak number actually running at once, and confirms every worker is a genuine virtual thread rather than a shared pool worker:

mapConcurrent(4, ...) results, completion order reversed: [10, 20, 30, 40]
CHECK ok   : mapConcurrent output preserves encounter order [10,20,30,40] even though task 1 finishes last
mapConcurrent(2, ...) over 8 tasks, peak concurrent in-flight = 2
CHECK ok   : peak in-flight never exceeded maxConcurrency=2, measured peak=2
CHECK ok   : the cap is actually reached, not just respected -- measured peak=2
VirtualThread[#38]/runnable@ForkJoinPool-1-worker-2 virtual=true
VirtualThread[#39]/runnable@ForkJoinPool-1-worker-2 virtual=true
CHECK ok   : every mapConcurrent worker thread reports isVirtual() == true
CHECK ok   : 6 elements produced 6 distinct virtual threads -- one per element, not a shared pool

Now the part worth slowing down for, still in MapConcurrentDemo.java. The natural assumption is that a mapper throwing an exception makes mapConcurrent behave like structured concurrency — cancel the siblings, fail fast. It does not:

Stream.of(1, 2, 3, 4, 5)
        .gather(Gatherers.mapConcurrent(5, n -> {
            if (n == 3) throw new IllegalStateException("boom at element 3");
            sleep(Duration.ofMillis(900));
            return n;
        }))
        .toList();
// throws IllegalStateException -- but not until ~900ms later

mapConcurrent propagated: boom at element 3 after 901ms
CHECK ok   : the failure only surfaced after 901ms -- mapConcurrent waited for the other 4 in-flight tasks, it did not cancel them

A failure doesn’t cancel the other tasks. Element 3’s mapper throws immediately; elements 1, 2, 4 and 5 each sleep 900ms. If mapConcurrent cancelled siblings on failure, the exception would surface in single-digit milliseconds. It doesn’t — it surfaces after ~900ms, once every already-started mapper has actually finished. If you’re using mapConcurrent for I/O-bound work and one call hangs, the whole terminal operation hangs with it, exception or not. StructuredTaskScope (see the companion post on Scoped Values, which covers it directly) is the tool that actually cancels siblings on failure — mapConcurrent is not that, despite living right next to virtual threads in the same release.

One or two paragraphs of reference-grade depth: the fixed concurrency cap makes mapConcurrent a natural fit for rate-limiting calls to a downstream service from inside a stream pipeline without reaching for a full executor and a Semaphore. The gap is that there’s no timeout parameter and no cancellation hook built in — both chapter 16 covers with a worked example of wrapping the mapper in your own deadline check.

• Going deeper: chapter 16 has the full failure-propagation timeline and a deadline-wrapping pattern for production use.
• Related: virtual threads and structured concurrency are covered end to end in the companion post on Scoped Values vs ThreadLocal.

Writing a Gatherer from scratch

Gatherer<T, A, R> is four composable pieces, and only one of them is required: an initializer supplying private mutable state of type A; an integrator that runs per element, can read and write that state, push zero or more R values downstream, and returns a boolean saying whether to keep pulling from upstream; a combiner for merging state across parallel splits (omit it — via Gatherer.ofSequential(...) — and the gatherer simply never runs in parallel); and a finisher that flushes anything left once the source is exhausted.

Both custom gatherers below live in CustomGatherer.java. First, dedupeConsecutive, which collapses adjacent equal elements — like Unix uniq, non-adjacent duplicates survive:

static <T> Gatherer<T, ?, T> dedupeConsecutive() {
    return Gatherer.ofSequential(
            () -> new ArrayList<Object>(List.of()),
            (state, element, downstream) -> {
                boolean changed = state.isEmpty() || !state.get(0).equals(element);
                if (changed) {
                    if (state.isEmpty()) state.add(element); else state.set(0, element);
                    return downstream.push(element);
                }
                return true;
            });
}

takeUntil is the case a Collector structurally cannot express: it stops pulling from upstream the moment its condition is met, by returning false from the integrator.

static <T> Gatherer<T, ?, T> takeUntil(Predicate<? super T> stopAfter) {
    return Gatherer.ofSequential((state, element, downstream) -> {
        downstream.push(element);
        return !stopAfter.test(element);   // false = stop; upstream not pulled again
    });
}

Source: CustomGatherer.java. The output proves the short-circuit is real — a peek() placed upstream of the gather only ever sees the elements actually consumed:

dedupeConsecutive of a,a,b,b,b,a,c,c = [a, b, a, c]
CHECK ok   : consecutive runs collapse to [a, b, a, c]; the second 'a' survives because it isn't adjacent to the first
takeUntil(n == 3) result = [1, 2, 3], upstream elements actually pulled = [1, 2, 3]
CHECK ok   : takeUntil includes the element that satisfies the stop condition: [1, 2, 3]
CHECK ok   : the upstream peek() only saw 1, 2, 3 -- elements 4, 5, 6 were never pulled, this genuinely short-circuits

Reference-grade note for anyone reaching for this on a truly large or infinite stream: because the integrator’s return value controls pulling, a custom gatherer is the mechanism for writing your own early-exit stream operation — something no combination of filter/limit/takeWhile can do once the stop condition depends on more than the current element.

• Going deeper: chapter 16 walks through writing a stateful top-N gatherer, and covers the combiner in full for anyone who does need parallel support.
• Reference: JEP 485’s own worked examples, including a distinctBy gatherer close in spirit to dedupeConsecutive above.

Collector vs Gatherer: when each is the right tool

A Collector always reduces to exactly one terminal value and can only sit at the very end of a pipeline. A Gatherer is an intermediate operation — it returns Stream<R>, so a gather can be followed by filter, map, another gather, or a collect. Reach for whichever one matches what you actually need at that point in the pipeline, not whichever one you reached for last time.

// The answer really is one Map -- a Collector is exactly right, a Gatherer would add nothing:
Map<String, Double> totals = orders.stream()
        .collect(Collectors.groupingBy(Order::customer, Collectors.summingDouble(Order::amount)));

// A running total that still needs filtering and mapping -- Collector would need a SECOND
// stream over an intermediate List; gather() keeps it one lazy pipeline:
List<String> overThreshold = Stream.of(5, 10, 15, 20, 25)
        .gather(Gatherers.scan(() -> 0, Integer::sum))
        .filter(total -> total > 20)
        .map(total -> "total=" + total)
        .toList();
// [total=30, total=50, total=75]

Source: CollectorVsGatherer.java. Output:

Collectors.groupingBy totals = {amy=19.75, bo=23.99}
CHECK ok   : amy's total via groupingBy/summingDouble is 19.75
scan -> filter -> map, one pipeline = [total=30, total=50, total=75]
CHECK ok   : scan feeding straight into filter/map in the same pipeline gives [total=30, total=50, total=75]
windowFixed(2) over 7 elements, then collect(counting()) = 4
CHECK ok   : collect(counting()) after gather() proves the gather result is still a Stream you can collect further: 4 windows

A short rule that holds up in practice: if the answer to “what does this pipeline produce” is genuinely one value, use a Collector. If the transformation needs to keep streaming afterward, needs private state across elements, or needs the ability to stop early, that’s a Gatherer — and the fact that gather()’s result is still a plain Stream means you can always collect() at the very end regardless of which one you used along the way.

• Going deeper: chapter 16 has a longer worked comparison including a case where switching a Collector-based pipeline to a Gatherer cut memory use by avoiding an intermediate List.
• Related: the Streams API and Collectors cookbook for the full Collector side of this comparison.

Should you reach for this?

Use the built-ins freely; write a custom Gatherer only when you actually need one. windowFixed, windowSliding, fold, scan and mapConcurrent replace real workarounds people have been writing by hand for years — reach for them the moment the built-in matches what you need. Writing your own Gatherer is the right call when you need genuine mid-stream state, a custom short-circuit, or output that isn’t 1-to-1 with input — but for a simple terminal reduction, a plain Collector is still less code and clearer intent. Don’t reach for gather() just because it’s new.

Further reading

• Companion repository: javademos, chapter 16 — Stream Gatherers
• Scoped Values vs ThreadLocal: Migration Guide with Virtual Threads — the companion post, covering StructuredTaskScope and real fail-fast cancellation
• Java 25 LTS: Every JEP That Matters — a wider roundup that touches JEP 485 alongside the rest of the 25 release; this post is the deep dive that roundup doesn’t have room for
• Official: JEP 485: Stream Gatherers
• Official Javadoc: java.util.stream.Gatherer, java.util.stream.Gatherers