Supercharging Java Streams with Custom Stream Gatherers (JDK 22+)

Java Streams, introduced back in Java 8, completely transformed how we process collections. We went from writing verbose for loops to clean, declarative pipelines using map(), filter(), and collect(). However, as any developer working on large-scale financial ledgers or data analytics in Bengaluru’s tech corridors knows, the standard stream API has a glaring limitation: its set of intermediate operations is fixed.

The Limitation of Traditional Streams

If you want to perform custom intermediate operations—like windowing, stateful filtering, or short-circuiting based on a custom condition—the stream API does not play nice.

For example, suppose you have a stream of transactions and want to group them into sliding windows of size 2 (e.g., comparing transaction N and N+1 to detect duplicate attempts within a short timeframe). With standard streams, you cannot easily do this without collecting everything to a list first (breaking the lazy evaluation benefits of streams) or using dangerous stateful hacks with mutable objects outside the stream (which breaks parallel stream safety and creates race conditions).

Traditional custom processing was restricted to terminal collectors (via Collector). But a collector runs at the very end of the stream, which means you cannot chain further stream operations after it.

Introducing Stream Gatherers (JEP 461)

To bridge this gap, JDK 22 introduced Stream Gatherers (JEP 461) as a preview feature. Gatherers allow you to define custom intermediate operations. The Stream interface now exposes a new .gather(...) method that accepts a Gatherer.

A Gatherer<T, A, R> is defined by three type parameters:

• T: The type of input elements.
• A: The type of private state container used during execution.
• R: The type of output elements emitted downstream.

To write a custom gatherer, you define four components:

1. Initializer (Supplier<A>): Creates the initial state container. If your gatherer is stateless, you can omit this.
2. Integrator (Integrator<A, T, R>): The core logic. It processes each element from the input stream, updates the state, and decides whether to emit an output element to the downstream or stop processing early (short-circuit).
3. Combiner (BinaryOperator<A>): Used for parallel streams. It defines how to merge two state containers.
4. Finisher (BiConsumer<A, Downstream<R>>): Invoked after the source stream is fully consumed. It is useful for emitting remaining state or performing cleanup.

Writing a Sliding Window Gatherer

Let’s implement a stateful sliding window gatherer that groups stream items into overlapping pairs of size 2. For example, a stream of [1, 2, 3, 4] should become [[1, 2], [2, 3], [3, 4]].

Here is the implementation:

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Gatherer;

public class SlidingWindowGatherer {

    public static <T> Gatherer<T, List<T>, List<T>> slidingWindow() {
        return Gatherer.ofSequential(
            // 1. Initializer: Create a list to store current window elements
            ArrayList::new,
            // 2. Integrator: Process each element
            (state, element, downstream) -> {
                state.add(element);
                if (state.size() == 2) {
                    // Emit a copy of the window
                    downstream.push(List.copyOf(state));
                    // Slide the window: remove the first element
                    state.remove(0);
                }
                return true; // Keep processing the stream
            }
        );
    }
}

Now, let’s see how we use this in a stream:

import java.util.List;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        List<List<Integer>> windows = Stream.of(1, 2, 3, 4, 5)
            .gather(SlidingWindowGatherer.slidingWindow())
            .toList();

        System.out.println(windows); 
        // Output: [[1, 2], [2, 3], [3, 4], [4, 5]]
    }
}

Why Gatherers Are a Game Changer

In this short example, the gatherer maintains a sliding state container (List<T>). When a new element comes in, it adds it to the list. Once the list has two elements, it pushes a copy to the downstream collector and removes the oldest element. This operates completely lazily—elements are processed one-by-one as they flow through the stream, without needing to buffer the entire stream in memory first!

Interview and Career Edge

If you are interviewing for senior Java roles at product startups or tech giants like Amazon India or Flipkart, mentioning JEP 461 will set you apart. Most candidates are stuck in the Java 8 world. Explaining how Stream Gatherers bring custom, stateful intermediate stream processing while preserving lazy evaluation and short-circuiting shows you are at the forefront of Java design patterns.