Parallel Pipeline That Loses Readings, Collapses Duplicates and Never Logs

ArrayList is not thread-safe, and add is not atomic. It reads size, writes elementData[size], then writes size + 1, and may call grow() first. With parallel forEach:

• Two threads can read the same size, both write index n, one value overwrites the other → lost updates (2.31M–2.40M instead of 2.4M), and size may end up lower than the number of writes.
• During grow(), the new array is assigned before or after the size bump depending on the interleaving, so a thread can index into the old, smaller array → ArrayIndexOutOfBoundsException.

forEach on a parallel stream is explicitly non-deterministic: it does not respect encounter order and runs the lambda concurrently. The lambda must be stateless and non-interfering.

Correct: readings.parallelStream().map(this::transform).collect(Collectors.toList()). A collector is built for parallel use: each thread accumulates into its own container and the containers are merged with a combiner, so there is no shared mutable state, and encounter order is preserved for an ordered source. Alternatives: reduce with an associative operator, Collectors.toConcurrentMap/groupingByConcurrent when a Map is the target, or Collections.synchronizedList (correct but a serialization bottleneck at 2.4M operations). forEachOrdered restores ordering but does nothing for thread safety.

Version B: the two-argument Collectors.toMap merges with a function that throws, so the first duplicate key produces IllegalStateException('Duplicate key ...'). The merge function receives (oldValue, newValue) and its return value replaces the existing one, so (a, b) -> b keeps the last value and (a, b) -> Math.max(a, b) state an explicit rule; either is acceptable only once you decide which feed wins, since the map is now order-dependent. Two extra traps: the two-argument toMap throws NullPointerException if any value is null (it delegates to Map.merge, which rejects null values), and it returns a HashMap, so add LinkedHashMap::new as the fourth argument if iteration order must follow the source. If duplicates are expected rather than exceptional, groupingBy(Reading::getSensorId, mapping(Reading::getValue, toList())) models it better.

Version C: unordered() is a hint that removes the encounter-order obligation. The spec says findFirst returns the first element, but if the stream has no encounter order, any element may be returned - it is effectively findAny, so results vary run to run. Fix: drop unordered() and keep findFirst (on a parallel stream it is still cheap; it reduces by index rather than materializing), or make the choice order-independent with min(Comparator.comparing(Reading::getTimestamp)), which yields the same answer in any order.

peek is an intermediate operation: it is invoked only for elements that actually flow through the pipeline, and only when the terminal operation pulls them. anyMatch is short-circuiting, so the pipeline stops as soon as a matching element is found: a sequential stream logs one line, and a parallel stream logs one line per worker thread that had already started, plus possibly a few buffered elements. Seeing 1–2 lines is the expected behaviour, not a logging bug.

A second, sharper trap: peek is not guaranteed to run at all for a given element. count() on a stream with a known size can be answered from the spliterator's estimateSize/getExactSizeIfKnown, skipping the pipeline entirely, so stream.peek(log::trace).count() may print nothing. Optimizations are free to elide peek because its contract says it exists mainly to support debugging.

So: peek is fine for temporary debugging during development; it is not a logging mechanism, and it must never mutate state that downstream stages depend on (that is interference). For real audit logging, put the side effect in a terminal operation (forEach(this::process)), or map to a small record and log in the collector, or log after collecting. If you need a per-record trace and a boolean answer, do two passes or collect first, then run anyMatch over the collected list.