Java Stream Reduction, Grouping, Partitioning and Gatherers (1Z0-831)

Stream reduction combines every element into one result, and Stream.reduce comes in three forms whose return types differ: with an identity it returns T, without one it returns Optional<T>, and the three-argument form returns a different type U.

import java.util.*;
import java.util.stream.*;

public class Main {
    public static void main(String[] args) {
        List<Integer> scores = List.of(10, 20, 12);
        Integer total = scores.stream().reduce(0, Integer::sum);
        Optional<Integer> best = scores.stream()
            .reduce(Integer::max);
        Optional<Integer> none = Stream.<Integer>empty()
            .reduce(Integer::max);
        List<Integer> asList = best.stream().toList();
        int chars = Stream.of("sun", "moon")
            .reduce(0, (acc, w) -> acc + w.length(), Integer::sum);
        System.out.println(total + " " + best + " " + none
            + " " + asList + " " + none.stream().toList()
            + " " + chars);
    }
}

This prints 42 Optional[20] Optional.empty [20] [] 7. Three consequences are easy to miss:

• The two-argument form cannot change type. Its BinaryOperator<T> needs both operands and the result to be the element type, so an accumulator that adds a number to a String element or to a record does not compile. Either map the elements to the result type first, or use the three-argument form, whose identity fixes U and whose combiner merges two partial U results.
• Identity-free forms give an Optional, not a value. On a Stream<T> the result is Optional<T>; on an IntStream it is OptionalInt. Neither converts to a plain value without get(), orElse() or getAsInt(). Optional.stream() turns the result back into a stream of zero or one element, so opt.stream().toList() is a one-element list when a value is present and an empty list when it is not.
• A null result is an error, not an empty Optional. If the accumulator of reduce(BinaryOperator) produces null as the final result, reduce throws NullPointerException — the Javadoc says so, because the result is wrapped in a non-null Optional. Optional.empty means only that the stream had no elements.

A parallel reduction gives the same answer as a sequential one only when three rules hold: the identity is a true identity (op(identity, x) == x for every x), the operation is associative, and the combiner agrees with the accumulator (combiner(u, accumulator(identity, t)) == accumulator(u, t)).

The reason is how parallel streams work: the source is split into chunks, each chunk is reduced starting from the identity, and the partial results are merged with the combiner (or the same operator in the two-argument form). The identity can therefore be applied once per chunk, and the number of chunks is not something your program controls.

Addition with identity 0 and the primitive streams' own sum() satisfy all three rules, which is why they are the safe default. The identity-free reduce(op) has no identity to misuse, but still needs an associative operation.

Choosing the identity means finding the value that never changes the result. For Integer::min that is Integer.MAX_VALUE; for Integer::max it is Integer.MIN_VALUE; for String concatenation it is ""; for multiplication it is 1. Picking 0 for a minimum silently replaces the smallest positive value with 0, even sequentially:

import java.util.*;
import java.util.stream.*;

public class Main {
    public static void main(String[] args) {
        List<Integer> waits = List.of(12, 30, 18);
        int right = waits.parallelStream()
            .reduce(Integer.MAX_VALUE, Integer::min);
        int wrong = waits.parallelStream().reduce(0, Integer::min);
        System.out.println(right + " " + wrong);
    }
}

This prints 12 0. When no neutral value exists, use the identity-free form and handle the Optional.

Mutable reduction accumulates elements into a container such as an ArrayList or StringBuilder, and collect(supplier, accumulator, combiner) does it safely even on a parallel stream because every chunk gets its own fresh container from the supplier.

The three functions have fixed jobs: the supplier creates an empty container, the accumulator adds one element to a container, and the combiner folds the second container into the first. collect keeps the first argument and discards the second, so a combiner must mutate its first argument. Collectors.toCollection(supplier) is the Collector form of the same mutable reduction: it creates containers with your supplier, adds with add and merges with addAll, with the same parallel safety and the same encounter-order guarantee — use it when you need a specific collection type such as LinkedList or TreeSet.

import java.util.*;
import java.util.stream.*;

public class Main {
    public static void main(String[] args) {
        List<String> colours = List.of("red", "green", "blue");
        ArrayList<Integer> lens = colours.parallelStream()
            .map(String::length)
            .collect(ArrayList::new, ArrayList::add,
                     ArrayList::addAll);
        LinkedList<String> linked = colours.parallelStream()
            .collect(Collectors.toCollection(LinkedList::new));
        String code = Stream.of("x1", "y2", "z3").parallel()
            .collect(StringBuilder::new, StringBuilder::append,
                     StringBuilder::append)
            .toString();
        System.out.println(lens + " " + linked + " " + code);
    }
}

This prints [3, 5, 4] [red, green, blue] x1y2z3 on every run. For an ordered source the framework combines partial results in encounter order, so order is preserved without any synchronisation. StringBuilder::append works as both accumulator (append(String)) and combiner (append(CharSequence), since a StringBuilder is a CharSequence).

A combiner is only called when the stream is actually split, so a broken one can look correct on a sequential run. It goes wrong in three ways: it ignores its arguments (the second chunk's elements vanish), it merges in the wrong direction (into the container that collect throws away), or it has the wrong shape — the combiner must be a BiConsumer<R, R> taking two containers, so a method that adds a single element does not fit and the call fails to compile.

Sharing one external list and filling it from forEach is the unsafe alternative this form replaces. The primitive streams (IntStream, LongStream, DoubleStream) have only this three-function collect; passing them a Collector such as Collectors.joining() does not compile until you call boxed() or mapToObj(...).

The terminal operations count, sum, average, max, min and summaryStatistics are ready-made reductions, and what each returns for an empty stream is part of its contract.

import java.util.*;
import java.util.stream.*;

public class Main {
    public static void main(String[] args) {
        System.out.println(IntStream.of(3, 6, 9).sum() + " "
            + IntStream.of(3, 6, 9).average() + " "
            + LongStream.empty().sum() + " "
            + DoubleStream.empty().average() + " "
            + IntStream.empty().max());
        IntSummaryStatistics st = IntStream.of(5, -3, 11)
            .summaryStatistics();
        System.out.println(st.getCount() + " " + st.getSum() + " "
            + st.getMin() + " " + st.getMax() + " "
            + st.getAverage());
        Comparator<String> byLen =
            Comparator.comparing(String::length);
        var trees = List.of("cedar", "yew", "spruce");
        System.out.println(trees.stream().min(byLen) + " "
            + trees.stream().min(byLen.reversed()) + " "
            + trees.stream().max(Comparator.naturalOrder()));
    }
}

The output is:

18 OptionalDouble[6.0] 0 OptionalDouble.empty OptionalInt.empty
3 13 -3 11 4.333333333333333

Optional[yew] Optional[spruce] Optional[yew]

summaryStatistics() collects count, sum, min, max and average in one pass and never throws: an empty one keeps the starting values of its min and max fields (the extreme opposite values, so any real element replaces them) and defines the average as 0.0. Stream.max and min on object streams take a Comparator and return the element that comparator ranks highest or lowest — so min with a reversed() comparator returns the longest string, and natural order ranks strings alphabetically, not by length. The Optional prints as Optional[value], not as the bare value.

Collectors supplies reductions you pass to collect: joining concatenates text, toMap builds a map from a key and value function, and teeing feeds every element to two collectors and merges their results.

joining has three forms: joining(), joining(delimiter) and joining(delimiter, prefix, suffix). The delimiter goes only between elements, so one element gets none; the prefix and suffix wrap the whole result exactly once and are present even when there are no elements. It is a Collector<CharSequence, ?, String>, so the stream's elements must be text: numbers or other objects need a map(String::valueOf) step first.

import java.util.*;
import java.util.stream.*;

public class Main {
    public static void main(String[] args) {
        String a = Stream.of("north", "south")
            .collect(Collectors.joining(" | ", "{", "}"));
        String b = Stream.<String>empty()
            .collect(Collectors.joining(";", "(", ")"));
        String c = List.of(1.5, 2.5).stream()
            .map(String::valueOf)
            .collect(Collectors.joining("/"));
        System.out.println(a + " " + b + " " + c);
    }
}

This prints {north | south} () 1.5/2.5.

toMap throws IllegalStateException ("Duplicate key …") when two elements produce the same key and no merge function is given. The three-argument form takes a merge function (existingValue, newValue) -> … — the value already in the map comes first, the newly arriving one second — and the four-argument form adds a map supplier such as TreeMap::new. If you need all values per key as a list, that is groupingBy, not toMap.

teeing(c1, c2, merger) runs both downstream collectors over the same elements and passes their results to the merger in that order. Its result is exactly what the merger returns, computed with the downstream result types: counting() yields Long, summingInt yields Integer, minBy/maxBy yield Optional. Arithmetic in the merger follows normal Java rules, so dividing an Integer by a Long is integer division and drops the fraction; averagingInt or a double operand gives the fractional mean.

import java.util.*;
import java.util.stream.*;

public class Main {
    public static void main(String[] args) {
        var byInitial = Stream.of("tea", "tonic", "gin")
            .collect(Collectors.toMap(
                s -> s.charAt(0),
                String::length,
                (kept, incoming) -> kept + incoming,
                TreeMap::new));
        System.out.println(byInitial);
        try {
            Stream.of("cat", "cow", "dog")
                .collect(Collectors.toMap(String::length,
                                          s -> s));
        } catch (IllegalStateException e) {
            System.out.println(e.getMessage());
        }
        String ratio = Stream.of(3, 8, 1, 6)
            .collect(Collectors.teeing(
                Collectors.summingInt(n -> n),
                Collectors.counting(),
                (sum, n) -> sum + "/" + n));
        double mean = Stream.of(3, 8, 1, 6)
            .collect(Collectors.averagingInt(n -> n));
        List<String> frozen = Stream.of("q", "p")
            .collect(Collectors.collectingAndThen(
                Collectors.toList(), Collections::unmodifiableList));
        int product = Stream.of(2, 3, 4)
            .collect(Collectors.reducing(1, (x, y) -> x * y));
        System.out.println(ratio + " " + mean + " " + frozen
            + " " + product);
    }
}

This prints:

{g=3, t=8}
Duplicate key 3 (attempted merging values cat and cow)
18/4 4.5 [q, p] 24

Two further wrappers are worth knowing: collectingAndThen(collector, finisher) transforms a collector's result (here into an unmodifiable list), and reducing(identity, op) is reduce in collector form, mainly useful as a downstream collector.

Collectors.groupingBy splits a stream into groups keyed by a classifier function and returns a Map whose value type is decided by the downstream collector — a List by default.

Downstream collectors determine the value type exactly, and a declaration that disagrees does not compile — the generic types are inferred from the collector, never converted:

• counting() → Long, so its map must be declared with Long values; a Map<K, Integer> target is an inference error. When Integer counts are needed, pick a collector that produces Integer.
• summingInt → Integer; summingLong → Long; summingDouble → Double
• averagingInt / averagingLong / averagingDouble → always Double
• mapping(mapper, downstream) → transforms each element, then applies the downstream (for example toList() or joining())
• minBy / maxBy → Optional<T>

import java.util.*;
import java.util.stream.*;

public class Main {
    record Trip(String mode, int km) {}

    public static void main(String[] args) {
        var trips = List.of(new Trip("bus", 12),
                            new Trip("rail", 40),
                            new Trip("bus", 8),
                            new Trip("rail", 25),
                            new Trip("walk", 2));
        Map<String, List<Trip>> byMode = trips.stream()
            .collect(Collectors.groupingBy(Trip::mode));
        TreeMap<String, Long> n = trips.stream()
            .collect(Collectors.groupingBy(Trip::mode,
                TreeMap::new, Collectors.counting()));
        Map<String, Double> avg = trips.stream()
            .collect(Collectors.groupingBy(Trip::mode,
                TreeMap::new,
                Collectors.averagingInt(Trip::km)));
        Map<String, Integer> sum = trips.stream()
            .collect(Collectors.groupingBy(Trip::mode,
                TreeMap::new,
                Collectors.summingInt(Trip::km)));
        Map<String, List<Integer>> kms = trips.stream()
            .collect(Collectors.groupingBy(Trip::mode,
                TreeMap::new,
                Collectors.mapping(Trip::km,
                                   Collectors.toList())));
        System.out.println(byMode.get("rail").size() + " " + n);
        System.out.println(avg + " " + sum);
        System.out.println(kms);
        var nested = Stream.of("kiwi", "lime", "lemon", "fig", "kale")
            .collect(Collectors.groupingBy(s -> s.charAt(0),
                TreeMap::new,
                Collectors.groupingBy(String::length,
                    TreeMap::new, Collectors.counting())));
        System.out.println(nested);
    }
}

The output is:

2 {bus=2, rail=2, walk=1}
{bus=10.0, rail=32.5, walk=2.0} {bus=20, rail=65, walk=2}
{bus=[12, 8], rail=[40, 25], walk=[2]}
{f={3=1}, k={4=2}, l={4=1, 5=1}}

Multilevel grouping is just a groupingBy (or partitioningBy) passed as the downstream: the outer classifier picks the first-level key, and each group is collected again by the inner one. A group only exists when at least one element maps to it — in the nested map above, k has no entry for length 5 and f none for length 4, because groupingBy never creates a key with a zero count. Values within a group keep encounter order.

Null keys are rejected. If the classifier returns null for any element, groupingBy throws NullPointerException ("element cannot be mapped to a null key"); it neither drops the element nor creates a null key. Map a missing value to a real key (such as "" or a constant) first.

Without a map factory the map implementation is unspecified, so its printed order is not something to rely on; pass TreeMap::new when order matters. groupingByConcurrent is the variant that builds a ConcurrentMap for unordered parallel collection.

Collectors.partitioningBy(predicate) splits a stream into exactly two groups and returns a Map<Boolean, …> that always contains both the false and true keys, even when one side is empty.

import java.util.*;
import java.util.stream.*;

public class Main {
    public static void main(String[] args) {
        var g = Stream.of("bus", "car")
            .collect(Collectors.groupingBy(s -> s.contains("z")));
        var p = Stream.of("bus", "car")
            .collect(Collectors.partitioningBy(s -> s.contains("z")));
        System.out.println(g + " " + p + " " + g.get(true));
        var avg = Stream.of(-4, 6, 10)
            .collect(Collectors.partitioningBy(n -> n > 0,
                Collectors.averagingInt(n -> n)));
        var cnt = Stream.of(-1, -2)
            .collect(Collectors.partitioningBy(n -> n > 0,
                Collectors.counting()));
        var names = Stream.of("Ola", "Bo", "Ines")
            .collect(Collectors.partitioningBy(s -> s.length() > 5,
                Collectors.joining("+")));
        System.out.println(avg + " " + cnt + " " + names);
    }
}

This prints:

{false=[bus, car]} {false=[bus, car], true=[]} null
{false=-4.0, true=8.0} {false=2, true=0} {false=Ola+Bo+Ines, true=}

The second form, partitioningBy(predicate, downstream), reduces each side with any collector. An empty side holds whatever that collector produces for zero elements — an empty list for the default, 0 for counting() or summingInt, an empty string for joining, Optional.empty for maxBy — never null. The partition map prints false before true. Used as the downstream of a groupingBy, it gives every group both keys, so each inner map has a false and a true entry even when one of them is empty.

Decomposition turns each element into zero or more elements — flatMap does it by returning a stream per element, mapMulti by pushing elements to a consumer — and concatenation joins two streams end to end with Stream.concat.

flatMap(mapper) takes a Function<T, Stream<R>>; the returned streams are flattened in order. If the mapper returns null, the documentation says an empty stream is used instead, so that element simply disappears. mapMulti(BiConsumer<T, Consumer<R>>) calls your code with each element and a sink; every accept emits one element. Because R appears only inside the consumer, javac often cannot infer it from a lambda and falls back to Object, so write a type witness: .<Integer>mapMulti(...). A method reference must match the expected shape too: a void method taking an element and a Consumer fits mapMulti, while flatMap needs a method that returns a stream.

Stream.concat(a, b) yields every element of a then every element of b, in argument order. It is lazy — elements are pulled from b only after a is exhausted and only if a downstream operation still wants more, so a short-circuiting terminal operation stops pulling as soon as it has its answer. Both arguments must be the same stream kind: Stream.concat takes two Streams, IntStream.concat two IntStreams, and mixing them does not compile. Streams are single-use, so passing one that was already consumed by a terminal operation, or already linked to another pipeline, throws IllegalStateException at the concat call itself.

import java.util.*;
import java.util.stream.*;

public class Main {
    public static void main(String[] args) {
        var words = Stream.of("to be", "or")
            .flatMap(s -> Arrays.stream(s.split(" ")))
            .toList();
        var digits = Stream.of("12", "3")
            .<Integer>mapMulti((s, sink) -> {
                for (char ch : s.toCharArray()) {
                    sink.accept(ch - '0');
                }
            })
            .toList();
        var lens = Stream.of("ab", "#x", "abc")
            .flatMap(s -> s.startsWith("#") ? null
                                           : Stream.of(s.length()))
            .toList();
        var nums = IntStream.concat(IntStream.range(0, 2),
                                    IntStream.of(9))
            .boxed().toList();
        var nested = Stream.of(List.of('a'), List.of('b', 'c'));
        var flat = nested.flatMap(List::stream).toList();
        System.out.println(words + " " + digits + " " + lens
            + " " + nums + " " + flat);
        Stream<String> src = Stream.of("k");
        src.map(String::length);
        try {
            Stream.concat(src, Stream.of("m"));
        } catch (IllegalStateException ex) {
            System.out.println(ex.getMessage());
        }
    }
}

This prints:

[to, be, or] [1, 2, 3] [2, 3] [0, 1, 9] [a, b, c]
stream has already been operated upon or closed

A stream of collections is not decomposed automatically: Stream.of(List.of('a'), List.of('b', 'c')) has two elements, both lists, and only flatMap(List::stream) flattens it.

A gatherer is a custom intermediate operation applied with Stream.gather(gatherer) (standard since Java SE 24), and java.util.stream.Gatherers supplies five ready-made ones: windowFixed, windowSliding, fold, scan and mapConcurrent.

import java.util.*;
import java.util.stream.*;

public class Main {
    public static void main(String[] args) {
        var src = List.of(10, 20, 30, 40);
        System.out.println(src.stream()
            .gather(Gatherers.windowFixed(3)).toList());
        System.out.println(src.stream()
            .gather(Gatherers.windowSliding(2)).toList());
        System.out.println(Stream.of(2, 3, 5)
            .gather(Gatherers.fold(() -> 1, (a, b) -> a * b))
            .toList());
        System.out.println(Stream.of("x", "y", "z")
            .gather(Gatherers.scan(() -> "", String::concat))
            .toList());
        System.out.println(Stream.of("b", "a", "c")
            .gather(Gatherers.mapConcurrent(2, String::toUpperCase))
            .toList());
        System.out.println(Stream.of(5, 6, 7)
            .gather(Gatherers.<Integer>windowFixed(2)
                .andThen(Gatherers.scan(() -> 0,
                    (acc, w) -> acc + w.getFirst())))
            .toList());
    }
}

The output is:

[[10, 20, 30], [40]]
[[10, 20], [20, 30], [30, 40]]
[30]
[x, xy, xyz]
[B, A, C]
[5, 12]

Three points decide most questions. First, the window gatherers emit Lists, so the result is a list of lists and a later map receives a whole window. Second, fold and scan are sequential gatherers with no combiner: even in a .parallel() pipeline the gather stage processes elements one at a time in encounter order, so the result is the same as on a sequential stream. Third, mapConcurrent exposes concurrency, not completion order — an element that takes longest to map still comes out in its original position. Gatherers compose with andThen: in the last line, windowFixed(2) produces [5, 6] and [7], and the scan adds each window's first element to a running total.

A custom gatherer is built with Gatherer.ofSequential(...) or Gatherer.of(...) from up to four functions: an optional initializer that creates private state, a required integrator called once per element, an optional combiner (only of takes one, enabling parallel evaluation), and an optional finisher run once at end of input.

• Integrator (state, element, downstream) -> boolean: it may push zero, one or many elements with downstream.push(x), and its return value says whether it wants more input. Returning false short-circuits the stream — later elements are never integrated. push itself returns false when the downstream wants no more, so returning the result of push propagates that.
• Order matters inside the integrator: Java evaluates && left to right, so when the push call comes before the stopping test, the element that triggers the stop has already been emitted. Test first and return false before pushing if that element must be dropped.
• Finisher (state, downstream) -> void: runs once when integration ends — after the last element, and also after the integrator returned false — and is where buffered leftovers or a summary must be pushed, or they are lost.
• Type Gatherer<T, A, R>: input type, state type (? or Void for stateless), output type. State is usually a mutable holder such as an array, a list or a small class, because the integrator cannot reassign it.

import java.util.*;
import java.util.stream.*;

public class Main {
    public static void main(String[] args) {
        Gatherer<String, String[], String> noRepeats =
            Gatherer.ofSequential(
                () -> new String[1],
                (last, s, down) -> {
                    if (s.equals(last[0])) {
                        return true;
                    }
                    last[0] = s;
                    return down.push(s);
                });
        System.out.println(Stream.of("a", "a", "b", "a")
            .gather(noRepeats).toList());

        Gatherer<Integer, int[], String> budget =
            Gatherer.ofSequential(
                () -> new int[1],
                (spent, n, down) -> {
                    if (spent[0] + n > 10) {
                        return false;
                    }
                    spent[0] += n;
                    return down.push("+" + n);
                },
                (spent, down) -> down.push("=" + spent[0]));
        System.out.println(Stream.of(4, 5, 3, 1)
            .gather(budget).toList());
    }
}

This prints [a, b, a] (the state remembers the last element pushed, so only consecutive repeats are dropped) and [+4, +5, =9] (adding 3 would pass the limit, so the integrator returned false, 1 was never seen, and the finisher still ran and pushed the total). A gatherer built with ofSequential is always evaluated sequentially, even in a parallel pipeline. Gatherer.of builds a parallelizable gatherer; for a stateful one you pass a combiner that merges two partial states, the same role a combiner plays in a parallel reduction.