Java Gatherers: Master Advanced Stream Pipelines With Custom Intermediate Operations

I once spent an entire afternoon rewriting a forty-line reporting loop into a stream pipeline, only to watch it run slower and fail three tests. The map and filter parts were lovely. The part where I needed to group every five records into a fixed-size batch, then flush whatever was left at the end, was not. I ended up with a Collector so baroque that a colleague asked me to add a comment explaining what the return type even meant. That was the moment I stopped pretending the Stream API could do everything.

Nothing about that story is unusual. Most Java developers hit the same wall. You learn map, filter, collect, reduce, and for a while those four feel like a complete toolkit. Then you need a sliding window, or a running total, or a dedupe of adjacent duplicates, and suddenly you are writing an anonymous class with a mutable List inside, hoping nobody runs it in parallel.

Gatherers exist because that wall is real. If you have ever written a custom Collector just to get intermediate output, or reached for a third-party library with a StreamEx or jOOλ import, this was built for you.

Before I show you the fun parts, let me be honest about when you should close this tab. If your pipeline is filter then map then toList, leave it alone. It is fast, obvious, and the JIT compiler loves it. If you are summing numbers, mapToInt and sum beat anything clever. If your logic needs two passes over the data, or needs to look at element n-1, then a stream is probably the wrong tool and a plain loop is the right one. I say this as someone who reflexively reaches for streams. The correct answer to “should I use a stream here?” is sometimes no.

Now, the good stuff.

A gatherer is a custom intermediate operation. That is the whole idea in one sentence. map is an intermediate operation that takes one element in and puts one element out. filter takes one in and puts zero or one out. A gatherer takes one element in and puts any number of elements out, at any time. Zero, one, five, or nothing for a while and then everything at once. And it can hold state between elements.

Every gatherer has the same four moving parts. There is an initializer, which creates the state object — think of it as a fresh scratchpad for each pipeline run. There is an integrator, which receives the scratchpad, the incoming element, and a downstream handle. The downstream handle is the doorway to the rest of the pipeline; you push results through it. The integrator returns a boolean that answers a simple question: do I want more elements?

The third part is a combiner, which only matters when you go parallel. And the fourth is a finisher, which runs once after the last element, so you can flush whatever is still sitting in your scratchpad.

That is it. Four parts. Once that clicks, every gatherer you will ever write is just a different scratchpad and a different reason to push.

Here is the smallest useful gatherer I know — one that drops repeated neighbours. If you have a log of retry attempts and want to collapse ERROR, ERROR, ERROR into a single ERROR, this does it.

import java.util.Objects;
import java.util.stream.Gatherer;

static <T> Gatherer<T, ?, T> dedupeConsecutive() {
    class State<T> {
        T last;
        boolean seen;
    }

    return Gatherer.of(
        State::new,
        (state, element, downstream) -> {
            if (!state.seen || !Objects.equals(state.last, element)) {
                state.last = element;
                state.seen = true;
                return downstream.push(element);
            }
            return true;
        }
    );
}

Read the integrator out loud. “If this element differs from the last one I kept, remember it and push it.” If it matches, silently return true — meaning “keep the elements coming, I just didn’t have anything to say about this one.” That is the entire mental model. The scratchpad holds one value, and the only decision is whether to push.

Now windows. Batching is the most common reason people go looking for gatherers, and there is a built-in for it. Gatherers.windowFixed(n) hands you immutable lists of exactly n elements, except possibly the last one.

List<List<Order>> batches = orders.stream()
    .gather(Gatherers.windowFixed(500))
    .toList();

batches.forEach(batch -> repository.insertAll(batch));

Compare that to the loop-based alternative, which involves an index, a nested subList call, and an off-by-one bug you will find in production. I have written that loop maybe thirty times in my career. I would be happy never to write it again.

The subtlety with windowFixed is the last batch. It can be short. If your downstream API rejects fewer than 500 rows, you need to handle it, and the way you handle it is by writing your own gatherer with a finisher. A finisher is just a callback that gets the scratchpad and the downstream handle one final time, after the source is exhausted.

static <T> Gatherer<T, ?, List<T>> batched(int size) {
    return Gatherer.of(
        ArrayList<T>::new,
        (batch, element, downstream) -> {
            batch.add(element);
            if (batch.size() == size) {
                List<T> out = List.copyOf(batch);
                batch.clear();
                return downstream.push(out);
            }
            return true;
        },
        (left, right) -> {
            left.addAll(right);
            return left;
        },
        (batch, downstream) -> {
            if (!batch.isEmpty()) {
                downstream.push(List.copyOf(batch));
            }
        }
    );
}

Notice List.copyOf before I clear. If I pushed the raw ArrayList and then cleared it, every batch I had already handed downstream would be empty by the time anything read it. That bug has eaten more of my evenings than I care to admit.

windowSliding is the other built-in, and it overlaps. With a window size of three over 1,2,3,4,5 you get [1,2,3], [2,3,4], [3,4,5]. That is exactly what you want for a moving average, a rolling checksum, or smoothing a noisy sensor feed.

record Reading(long timestamp, double celsius) {}

List<Double> smoothed = readings.stream()
    .gather(Gatherers.windowSliding(5))
    .map(window -> window.stream()
        .mapToDouble(Reading::celsius)
        .average()
        .orElseThrow())
    .toList();

The mapToDouble inside is doing real work. A five-element window iterated as a List<Reading> and boxed into Double values is fine for a few thousand readings and wasteful for a few million. Keep the per-window work primitive. I will come back to this.

There are two more built-ins. Gatherers.scan produces every intermediate accumulation — a running total, a prefix sum — which is the thing reduce cannot give you because it only emits the final answer.

List<Integer> runningTotals = numbers.stream()
    .gather(Gatherers.scan(() -> 0, Integer::sum))
    .toList();

Give it 1,2,3,4 and you get 1,3,6,10. One line, and it replaces a loop that would have had an accumulator variable declared outside it, which is exactly the kind of escaping state that makes code hard to test.

Gatherers.fold is the same shape but emits only the final value. It reads like reduce but with a mutable seed, which is genuinely useful when your accumulator is a StringBuilder or an ArrayList and rebuilding it on every step would be quadratic.

Now the pattern I reach for most, and the one that justifies the whole feature: early termination. A gatherer’s integrator can return false, which tells the pipeline “I am done, stop feeding me.” That is a real short-circuit, not a filter that keeps scanning the rest of an infinite source.

static <T> Gatherer<T, ?, T> takeFirstMatching(Predicate<T> predicate, int limit) {
    class State { int found; }

    return Gatherer.of(
        State::new,
        (state, element, downstream) -> {
            if (predicate.test(element)) {
                state.found++;
                if (!downstream.push(element)) {
                    return false;
                }
                return state.found < limit;
            }
            return true;
        }
    );
}

Two different false returns, and they mean opposite things. The inner one says the downstream is finished — maybe the consumer called findFirst() or hit a limit. The outer one says this gatherer is finished. Getting that distinction right is most of the work in writing a correct short-circuiting gatherer.

Time-boxed batching is the pattern that finally got me to stop writing custom Collector classes. You want to push rows to an API every two seconds or every hundred rows, whichever comes first.

static <T> Gatherer<T, ?, List<T>> batchedByTime(int maxSize, Duration maxWait) {
    class State {
        final List<T> items = new ArrayList<>();
        long openedAt = System.nanoTime();
    }

    long maxNanos = maxWait.toNanos();

    return Gatherer.of(
        State::new,
        (state, element, downstream) -> {
            state.items.add(element);
            boolean full = state.items.size() >= maxSize;
            boolean expired = System.nanoTime() - state.openedAt >= maxNanos;
            if (full || expired) {
                List<T> out = List.copyOf(state.items);
                state.items.clear();
                state.openedAt = System.nanoTime();
                return downstream.push(out);
            }
            return true;
        },
        (batch, downstream) -> {
            if (!batch.items.isEmpty()) {
                downstream.push(List.copyOf(batch.items));
            }
        }
    );
}

I run this on sequential streams only, and I would encourage you to do the same. Time and parallelism do not mix politely. The combiner is deliberately absent because merging two half-full time windows has no obvious correct answer.

That leaves concurrency. Gatherers.mapConcurrent(n, fn) runs your mapping function on up to n threads at once while preserving the original order downstream. This is the gatherer I was most sceptical about and now use constantly for I/O-bound work.

List<Profile> profiles = userIds.stream()
    .gather(Gatherers.mapConcurrent(16, profileClient::fetch))
    .toList();

Sixteen in-flight HTTP calls, results in input order. Every CompletableFuture choreography I have written to do this took fifty lines and had a race condition in it.

Ten patterns is a lot to remember, so I package mine. I keep a single Pipelines class with static factory methods, and I compose gatherers the same way I compose map and filter — by chaining .gather() calls. A stream is just a pipeline; gatherers are stages in it.

List<Invoice> result = raw.stream()
    .filter(Objects::nonNull)
    .gather(Pipelines.dedupeConsecutive())
    .gather(Pipelines.batchedByTime(200, Duration.ofSeconds(2)))
    .flatMap(List::stream)
    .gather(Pipelines.batchFully(50))
    .flatMap(List::stream)
    .toList();

The .gather() calls are ordinary intermediate operations, so laziness holds. Nothing runs until that final toList(). And the kit is testable in isolation, which matters more than it sounds.

Now the part where I stop being enthusiastic.

Every gatherer adds a layer of indirection, and layers cost. The Stream machinery already boxes aggressively, and a gatherer that pushes List<T> wrappers around small groups makes that worse. If you are processing numbers, use IntStream, LongStream, or DoubleStream wherever the shape allows. Then measure.

Measure with JMH, not with System.currentTimeMillis(). Fifteen warm-up iterations, fifteen measurement iterations, @Fork(3), and @BenchmarkMode(Mode.AverageTime). Anything less and you are measuring the JIT compiler’s mood.

@State(Scope.Benchmark)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Warmup(iterations = 15)
@Measurement(iterations = 15)
@Fork(3)
public class PipelineBenchmark {

    private List<Integer> data;

    @Setup
    public void setup() {
        data = IntStream.range(0, 1_000_000).boxed().toList();
    }

    @Benchmark
    public long loopSum() {
        long total = 0;
        for (int value : data) {
            total += value;
        }
        return total;
    }

    @Benchmark
    public int streamSum() {
        return data.stream().mapToInt(Integer::intValue).sum();
    }

    @Benchmark
    public long gatherBatchedSum() {
        return data.stream()
            .gather(Pipelines.batchFully(1000))
            .mapToLong(batch -> batch.stream().mapToInt(Integer::intValue).sum())
            .sum();
    }
}

I have run benchmarks like this many times. The loop usually wins on raw arithmetic. That is not a strike against streams; it is a reminder that mapToInt exists for a reason, and that a pipeline which boxes a million Integer objects and then unboxes them again is doing work nobody asked for.

When a pipeline is slower than you expect, the question is rarely “is the stream slow?” The useful questions are: how many objects am I allocating per element, and where are they going? Turn on -XX:+UnlockDiagnosticVMOptions with JFR, or run with -prof gc in JMH, and look at allocation rate per operation rather than wall-clock time. Boxing inside a hot loop shows up immediately as gigabytes per second of garbage. I have found more real problems by reading allocation profiles than by reading flame graphs.

Short-circuiting deserves its own warning. findFirst, anyMatch, limit, and takeWhile are all short-circuit operations, and they stop pulling from the source as soon as the answer is known. That works beautifully with lazy sources and infinite generators. It does not work with sorted(), because sorting must consume everything before it produces one element. Put your limit before your sorted when you can, and your early termination will actually terminate early.

Parallel streams are the other trap, and it is a big one. Every parallel stream in your JVM, by default, shares a single ForkJoinPool.commonPool() sized to your processor count minus one. That pool has a hard limit and no queue-depth control.

// This looks harmless. Under load, it is not.
List<Result> results = items.parallelStream()
    .map(item -> httpClient.call(item))   // blocks for 100ms+
    .toList();

If you have sixteen cores and a hundred blocking calls, fifteen of them run at a time and the rest wait. Worse, anything else in your application that uses the common pool — a parallel sort, a CompletableFuture with no explicit executor — is now queued behind your network calls. I have watched this happen in production. The symptom was a slow sort in an unrelated part of the service.

The fix is either Gatherers.mapConcurrent, which bounds concurrency explicitly and does not squat on the common pool, or a dedicated Executor with ForkJoinPool or Executors.newFixedThreadPool. Pick a concurrency limit based on what the downstream service can take, not on how many cores you have. Cores are irrelevant when you are waiting on a socket.

There is one more honest question to ask: should this be a loop at all? My rule is that if the logic needs the previous element, the next element, two passes, or mutable state that outlives a single accumulation, a stream is fighting the problem. Gatherers push that boundary a long way out — you can now keep state legitimately — but a for loop with a comment above it is often clearer and faster than a gatherer with a combiner nobody can reason about. I have deleted gatherers in code review for exactly this reason. Mine, mostly.

Testing a gatherer is easier than testing a Collector, and I am grateful for that. A gatherer is a pure factory. Call it, feed it a stream, assert on the output.

@Test
void batches_everything_and_flushes_the_tail() {
    List<Integer> input = List.of(1, 2, 3, 4, 5, 6, 7);

    List<List<Integer>> result = input.stream()
        .gather(Pipelines.batchFully(3))
        .toList();

    assertThat(result).containsExactly(
        List.of(1, 2, 3),
        List.of(4, 5, 6),
        List.of(7)
    );
}

@Test
void dedupe_collapses_adjacent_duplicates_only() {
    List<String> input = List.of("a", "a", "b", "a", "a");

    List<String> result = input.stream()
        .gather(Pipelines.dedupeConsecutive())
        .toList();

    assertThat(result).containsExactly("a", "b", "a");
}

The second test is the one people get wrong. Adjacent dedupe is not distinct(). distinct() would give you a, b. If you meant distinct(), use distinct() — it is simpler and it parallelises correctly.

That last point generalises. Before writing a gatherer, check the built-ins. Gatherers.windowFixed, windowSliding, scan, fold, and mapConcurrent cover a surprising amount of ground, and they are tested, documented, and tuned. Write your own only when the built-in genuinely does not fit.

One more thing I wish someone had told me earlier: when you do write your own, decide up front whether it supports parallel execution, and say so in the code. If your combiner is a stub that throws UnsupportedOperationException, that is a valid and honest design — it tells the next person not to try. A gatherer whose combiner silently produces wrong results in parallel is a bug waiting for a Tuesday afternoon. And mark your integrator greedy with Integrator.ofGreedy when you know it consumes every element it is handed, because the runtime can schedule that more freely.

I have been writing Java streams since they arrived, and for most of that time my advice about custom intermediate operations was “don’t, unless you must.” Gatherers changed that. Batching, windowing, deduplication, running totals, bounded concurrency, and early termination now take a handful of lines each, and they compose into pipelines I can actually read six months later.

They arrive properly in Java 24. If you are on 22 or 23 you will need --enable-preview, which means you should not ship it. That is fine. Use the time to rewrite one ugly Collector into a gatherer, then write a JMH benchmark around it, then look at the allocation numbers. That loop — write, measure, decide — is the whole job. The gatherer is just the newest tool in it, and the only one that ever made me glad I had not given up on streams.