Stateful lambdas and side effects

Key points

• A lambda passed to map or filter must be stateless - its result may depend only on its input. Touching outside state makes the outcome depend on execution order.
• Sequentially this works and hides the bug. Switch to parallel and the same code loses elements, duplicates them, or throws - and it does so intermittently, which is worse than failing outright.
• Adding to a plain ArrayList from forEach is the common form. It is not thread-safe, so a parallel run can produce a shorter list, nulls, or an ArrayIndexOutOfBoundsException.
• Using a concurrent collection removes the crash but not the design problem: the result order becomes nondeterministic, and you have reimplemented collect badly.
• collect exists precisely for this. It uses a supplier, accumulator and combiner so each thread works on its own container and results merge safely.
• peek is for debugging, not for work. It may be skipped entirely when the pipeline can prove the element is not needed, so side effects inside it are not guaranteed to run at all.

Example

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class StatefulLambdas {

    public static void main(String[] args) {
        int n = 100_000;

        // WRONG - not thread-safe. Sequentially it looks fine.
        List<Integer> unsafe = new ArrayList<>();
        IntStream.range(0, n).forEach(unsafe::add);
        System.out.println("sequential forEach into ArrayList : " + unsafe.size()
                + "  (correct, and hides the bug)");

        // The SAME code in parallel: lost elements, or an exception.
        List<Integer> broken = new ArrayList<>();
        try {
            IntStream.range(0, n).parallel().forEach(broken::add);
            System.out.println("parallel   forEach into ArrayList : " + broken.size()
                    + (broken.size() == n ? "  (got lucky this run)" : "  <- LOST ELEMENTS"));
        } catch (Exception e) {
            System.out.println("parallel   forEach into ArrayList : "
                    + e.getClass().getSimpleName() + "  <- corrupted internals");
        }

        // Thread-safe but slow, and the ORDER is nondeterministic.
        List<Integer> cow = new CopyOnWriteArrayList<>();
        IntStream.range(0, 2_000).parallel().forEach(cow::add);
        System.out.println("parallel into CopyOnWriteArrayList: " + cow.size()
                + "  (safe, but order is arbitrary and it is O(n) per add)");

        // CORRECT - collect handles the merging for you.
        List<Integer> correct = IntStream.range(0, n).parallel().boxed()
                .collect(Collectors.toList());
        System.out.println("parallel collect(toList)          : " + correct.size()
                + "  and order preserved: " + (correct.get(0) == 0 && correct.get(n - 1) == n - 1));

        // A stateful FILTER - the result depends on evaluation order.
        AtomicInteger seen = new AtomicInteger();
        long everySecond = IntStream.range(0, 1_000).parallel()
                .filter(i -> seen.getAndIncrement() % 2 == 0)   // stateful: WRONG
                .count();
        System.out.println();
        System.out.println("stateful filter kept " + everySecond
                + " of 1000 (expected 500, but not reproducible)");

        // peek is NOT guaranteed to run - the pipeline may skip elements.
        AtomicInteger peeked = new AtomicInteger();
        long counted = IntStream.range(0, 1_000).boxed()
                .peek(i -> peeked.incrementAndGet())
                .count();
        System.out.println("count() = " + counted + " but peek ran " + peeked.get()
                + " times  <- count can skip the pipeline entirely");
    }
}

A stateful lambda is correct sequentially and intermittently wrong in parallel - use collect, which merges safely by design.