Order Totals: Duplicate Keys, Ordering and a Silent Race

The two-argument toMap(keyMapper, valueMapper) collects into a HashMap and, on a duplicate key, calls the underlying merge with no remapping function - it deliberately fails with IllegalStateException instead of silently overwriting, because both values are legitimate and the API will not guess. The unit test passed only because it happened to use four distinct customers.

TreeMap<String, BigDecimal> totals = orders.stream()
    .collect(Collectors.toMap(
        Order::customerId,
        Order::amount,
        BigDecimal::add,   // merge: (existingValue, newValue) -> merged
        TreeMap::new));    // map supplier

Overloads: the 2-arg form has no merge and no supplier; the 3-arg form adds a merge function but still produces a HashMap; the 4-arg form adds the supplier. The merge function receives the value already in the map first and the new value second, which matters for non-commutative merges such as string concatenation.

new TreeMap<>(map) also sorts, but it is a second pass, it re-hashes nothing while copying, and if the pipeline is later refactored to return Map from a method declaring an unordered map, the ordering guarantee silently disappears. Also note toMap throws NullPointerException when the value mapper returns null, so filter or map null amounts first.

ArrayList is not thread-safe. Concurrent add calls race on size and on the internal array: two threads can read the same size, both write into the same slot, and one element is silently lost - exactly the "short by three or four" symptom, with no exception because the array rarely needs to grow at the same moment. Additionally, forEach on a parallel stream does not preserve encounter order, so even a thread-safe collection would produce an arbitrary ordering.

Correct approach: never mutate shared state from a stream lambda. Let the framework do the accumulation, which is designed to combine per-thread partial results:

BigDecimal total = orders.parallelStream()
    .map(Order::amount)
    .reduce(BigDecimal.ZERO, BigDecimal::add);

List<BigDecimal> amounts = orders.parallelStream()
    .map(Order::amount)
    .collect(Collectors.toList());      // or toConcurrentMap for a map keyed by id

Collectors.toList() is safe in parallel because it uses a combiner internally. Three traps worth naming: forEachOrdered fixes ordering but not the data race; a reduce with a mutable identity such as new ArrayList<>() is a bug because the same instance is reused across threads; and a ConcurrentHashMap-backed collector still needs a proper merge function.

peek is an intermediate operation explicitly documented as being intended for debugging, and there is no guarantee that it is invoked. Since Java 9, count() is allowed to compute the result directly from the spliterator when the stream is SIZED and no intermediate operation changes the number of elements: an unfiltered orders.stream().peek(...).count() can return spliterator().getExactSizeIfKnown() and skip the pipeline entirely, so the consumer never runs.

The inconsistency that confuses people: add .filter(o -> o.amount.signum() > 0) and the size is no longer known for free, so count() walks the elements and the same peek starts printing. Neither behaviour is a bug; both are permitted by the specification.

To observe progress, use a terminal operation that must consume every element, or a collector that counts as a side effect of traversal:

long n = orders.stream()
    .peek(o -> log.debug("seen {}", o.customerId()))
    .collect(Collectors.counting());

For real diagnostics prefer logging inside the mapping stage or a profiler. Never put business side effects in peek; on a parallel stream they also run out of order and on arbitrary threads.