Ordering and Duplicates in a Parallel Grouping Pipeline
A log aggregator reads 200,000 lines and builds Map<String, List<Event>> grouped by tenant with Collectors.groupingBy(Event::tenant). Two defects appear. First, a de-duplication step using Collectors.toMap(Event::id, e -> e) fails on retried events with IllegalStateException: Duplicate key. Second, the generated report lists tenants in a different order on every run, breaking a golden-file comparison. The proposed fix - collecting into a HashMap inside a parallelStream() with forEach(map::put) - intermittently loses entries and once threw ArrayIndexOutOfBoundsException from HashMap.put. Requirements: de-duplicate by event id keeping the newest timestamp, emit tenants sorted by id, and within each tenant sort events by timestamp.
1. Why does `Collectors.toMap(Event::id, e -> e)` throw on duplicate keys, and what is the minimal correct replacement?
Read the worked answer
The two-argument toMap uses a throwing merge: when the merge function is omitted it is effectively (a, b) -> { throw new IllegalStateException("Duplicate key ..."); }. It also silently returns a HashMap, which is where the unstable ordering comes from.
Supply a merge function that resolves the conflict by business rule:
Collectors.toMap(
Event::id,
e -> e,
(a, b) -> a.timestamp().isAfter(b.timestamp()) ? a : b,
TreeMap::new); // sorted keys, stable output
Points worth stating: the four-argument overload lets you pick the map implementation; toMap rejects null values (it uses Map.merge), so a nullable value needs HashMap::new plus a manual put or a non-null sentinel. If the key is not actually unique by design, groupingBy is the honest collector, not toMap with a merge hack.
2. Why is collecting into a shared `HashMap` with `forEach(map::put)` broken under `parallelStream()`, even though `HashMap.put` is "just one call"?
Read the worked answer
HashMap is not thread-safe. Two threads calling put concurrently can both observe the same bucket, resize concurrently, and corrupt the table - the symptoms are exactly the ones described: dropped entries, a spurious ArrayIndexOutOfBoundsException from a resize, or (in older JDKs) a spinning get.
Collectors are built for this: each parallel chunk accumulates into its own container and the containers are merged at the end by code that respects happens-before. groupingBy, toMap, and groupingByConcurrent are safe; a shared mutable target is not.
Second, independent issue: forEach on a parallel stream does not preserve encounter order - forEachOrdered does. That is irrelevant for a HashMap sink but critical if you ever write to an ordered structure or an output file.
A synchronized block around the put would be correct but serialises the whole pipeline and throws away the parallelism.
3. Write the full pipeline that meets the stated requirements, and note any trap in reusing the stream.
Read the worked answer
Map<String, List<Event>> byTenant = events.parallelStream()
.collect(Collectors.groupingBy(
Event::tenant,
TreeMap::new, // tenants sorted by id
Collectors.collectingAndThen(
Collectors.toMap(Event::id, e -> e,
(a, b) -> a.timestamp().isAfter(b.timestamp()) ? a : b),
m -> m.values().stream()
.sorted(Comparator.comparing(Event::timestamp)
.thenComparing(Event::id))
.toList())));
Reasoning: the TreeMap supplier fixes key order; collectingAndThen de-duplicates before sorting, so the sort cost is paid on fewer elements; the secondary comparator on id makes ties deterministic, which is what a golden-file test actually requires. toList() returns an unmodifiable list.
Traps: a stream is consumed by its terminal operation - a second terminal call throws IllegalStateException: stream has already been operated upon or closed, so build both the grouped view and the flat de-duplicated view from the source collection, not from one stream. Also, HashMap.values() has no order, which is harmless only because sorted() is applied afterwards.