Parallel Side Effects in a 1.2 Million Row Trade Report

ArrayList is not thread-safe. add does an unsynchronized elementData[size++] = e, so concurrent writers overwrite each other's slots and lose size increments; a resize can also expose a partially copied array and throw ArrayIndexOutOfBoundsException. Nothing in the stream API makes forEach atomic or ordered for a parallel pipeline, so the race is the program's bug, not a stream quirk. Even Collections.synchronizedList would not help here if the list is later iterated without external synchronization.

Mutating shared state from a lambda is the anti-pattern. Collect instead, letting the framework handle per-thread accumulation and merging:

List<String> out = trades.parallelStream()
    .filter(t -> t.getAmount() > 1000)
    .map(Trade::getSymbol)
    .collect(Collectors.toList());

This is correct for parallel execution: each task accumulates into its own container and the partial results are combined, and for an ordered stream the encounter order is preserved. Use Collectors.toCollection(ArrayList::new) if a specific mutable type is needed, and Collectors.toUnmodifiableList() if the result should not change afterwards.

If a counter is genuinely needed, use LongAdder (preferred under contention) and attach it to a terminal operation whose result you actually use; AtomicInteger is correct but serializes on one cache line.

The two-argument Collectors.toMap uses Map.merge semantics with no merge function, so when a second value arrives for an existing key it throws IllegalStateException naming the duplicate key and both values. It is a deliberate fail-fast default, not an edge case.

The three-argument form supplies the merge function:

Map<String, Trade> largest = trades.stream().collect(Collectors.toMap(
    Trade::getSymbol,
    t -> t,
    (a, b) -> a.getAmount() >= b.getAmount() ? a : b));

Caveats worth stating in review:

• toMap rejects null values with a NullPointerException (it funnels through Map.merge), so filter or map nulls away first.
• A merge function that returns null removes the entry - safe as long as the stream has no null values, otherwise the NPE above applies.
• Ordering: the default map is a HashMap, so iteration order is unspecified. Supplying LinkedHashMap::new gives insertion order for a sequential stream, but under parallelStream() each task builds its own partial map and the combiner appends them, so insertion order is not encounter order. If deterministic order matters, collect sequentially or sort the resulting entries afterwards by a defined key.
• groupingBy(Trade::getSymbol, Collectors.maxBy(Comparator.comparing(Trade::getAmount))) expresses the same intent and returns Optional values, which is clearer when a symbol might have ties or when the merge rule may later change.

findFirst promises the first element in encounter order. Encounter order is only defined by the source: a List, a sorted stream, a LinkedHashSet. A ConcurrentHashMap.keySet() spliterator is unordered and weakly consistent - it may reflect updates made after the stream was created, may miss elements, and defines no encounter order at all. So findFirst is faithfully returning the first element of an order that the source never specified, and that order shifts as the ingestion thread mutates the map. The fix is not in the terminal operation: use a source with a defined order, or sort explicitly (e.g. by trade id) if the reported record must be reproducible.

The peek count has two independent causes. First, peek is an intermediate operation executed only for elements that are actually consumed downstream; findFirst is short-circuiting, so processing stops once a match is found in encounter order and each task only peeks its processed prefix - 37,000 of 1.2 million is entirely plausible. Second, peek is documented as being intended mainly for debugging: its visibility depends on the pipeline, and a shared side effect inside it is unreliable evidence.

Use findAny when the business rule is genuinely any matching trade (e.g. a sanity check that at least one exists). It frees parallel tasks to return the first element any thread finds, avoiding the ordering work findFirst must do, and it correctly signals that the choice is arbitrary. Do not use findAny merely to silence a nondeterminism bug you actually care about.