Java SE 8 Programmer II: A Parallel Stream That Loses Rows and Starves the Common ForkJoinPool

A Parallel Stream That Loses Rows and Starves the Common ForkJoinPool

The analytics team rewrote the nightly report from loops to streams: orders.parallelStream().filter(o -> o.total() > 100).sorted(comparing(Order::date)).forEach(o -> results.add(o)), where results is a plain ArrayList. On a 16-core box with 120,000 orders, roughly 1,100 rows vanish per run and the list occasionally contains nulls. In the web tier a controller uses .parallelStream() for a per-request lookup of 40 items; under 30 concurrent requests p99 latency rose from 40 ms to 900 ms, and an unrelated background task that also uses the common pool stalled for minutes. Re-running the report on identical input produced different row order. One developer wrapped a checked IOException in RuntimeException inside map, and another wants to use peek both to warm a cache and to trigger a per-element database write.

1. Explain why rows disappear and nulls appear, and give the corrected pipeline. Would `Collections.synchronizedList` fix it?

Read the worked answer

Three defects compound:

ArrayList is not thread-safe. Parallel forEach calls add from many threads; size++, the elementData[size] = e write and the grow/Arrays.copyOf step are each non-atomic, so writes are lost and stale or null slots become visible. That explains the missing ~1,100 rows and the nulls.
forEach on a parallel stream is explicitly non-deterministic in encounter order, even when the upstream is sorted().
There is no happens-before between the worker threads and the reader, so even successfully added elements are not guaranteed visible without a proper terminal operation.

Corrected pipeline: List<Order> out = orders.parallelStream().filter(o -> o.total() > 100).sorted(comparing(Order::date)).collect(Collectors.toList()); - collect gives each task its own accumulator, combines them safely, and preserves encounter order for an ordered stream. Alternatives: toCollection(ArrayList::new), toConcurrentMap, or groupingByConcurrent if a map is the real target.

Collections.synchronizedList is not a fix: iteration and combination still are not atomic relative to the pipeline, and every add serializes on one monitor, so you pay parallel overhead and still get contention. It converts a data race into a correctness-by-luck scenario that will break again on the next refactor.

2. Why did the 40-item lookup slow down so badly under concurrency, and how should the team decide between sequential and parallel execution?

Read the worked answer

The common ForkJoinPool is one JVM-wide pool with parallelism availableProcessors() - 1, i.e. 15 here. Two consequences:

Sharing. Every parallelStream() in the JVM submits to that same pool, including the controller and the unrelated background task. The background task stalled because controller tasks occupied the workers; conversely a blocking task anywhere in the JVM starves everything else.
Overhead. Splitting 40 elements into fork/join tasks, coordinating, and merging costs far more than 40 trivial lookups. A jump from 40 ms to 900 ms under 30 concurrent requests is consistent with queueing and context switching, not with computation.

Decision rule: keep streams sequential unless the data set is large, per-element work is non-trivial (CPU-bound, no I/O), and the source splits cheaply - arrays and ArrayList do; LinkedList, Stream.iterate and most Iterator sources do not. Never do blocking I/O inside a parallel stream unless you use ManagedBlocker, because the pool has no spare threads to compensate.

If parallelism is genuinely required inside a request, submit the work to a dedicated, bounded ForkJoinPool or a separate ExecutorService so it cannot starve unrelated work. Benchmark sequential versus parallel on realistic data before merging; the safe default is sequential.

3. The report must emit rows in date order, and two side-effecting ideas are on the table. Address ordering guarantees, the `peek` misuse, and the checked-exception handling.

Read the worked answer

sorted() produces an ordered stream, but forEach is documented as non-deterministic for parallel streams and does not respect encounter order. That is why identical inputs gave different orderings. Use forEachOrdered only when order is mandatory and collecting is not an option - it effectively serializes the tail of the pipeline - or simply collect(toList()), which preserves encounter order for ordered streams and lets the caller iterate in order.

peek is specified for debugging: it is an intermediate operation whose execution may be optimized away (in JDK 8, count() can skip the whole pipeline when the size is known), so using it to warm a cache or write to the database is a correctness bug that only appears at some call sites. Put side effects in the terminal operation, or map to a value and act on the collected result; for one write per element a plain loop with try-with-resources is clearer and easier to retry.

The checked exception is the other trap: Function.apply cannot throw a checked exception, and wrapping in RuntimeException is only acceptable if the caller unwraps and preserves the cause - otherwise the IOException is buried and error handling degrades. Wrap in a domain exception carrying the cause and the element, or keep that step outside the stream. Do not ship 'sneaky throw' helpers in production code; they defeat the compiler's checked-exception guarantee for every caller up the stack.