The report that slowed down everything else

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.stream.*;

/**
 * Solution: the report that slowed down everything else.
 *
 * Three defects, one per parallelStream, and they are three different
 * mistakes rather than three instances of one.
 *
 *   Defect 1 — blocking IO on the common ForkJoinPool. parallelStream does
 *   not create threads; it submits to a JVM-wide singleton whose parallelism
 *   is cores minus one. Forty calls sleeping for 5ms each occupy every worker,
 *   and any other parallel stream in the process — including inside a library
 *   — queues behind them. That is why unrelated endpoints timed out while a
 *   report ran, and why no benchmark of the report alone showed it.
 *
 *   The fix is a pool this job owns, sized for the remote service rather than
 *   for the CPU. The concurrency limit becomes a decision instead of a side
 *   effect of the core count, and nothing else in the JVM is affected. On 21
 *   a virtual-thread executor is the better answer again, since the work is
 *   waiting rather than computing.
 *
 *   Defect 2 — a shared ArrayList in a parallel forEach. Elements are lost and
 *   the list can throw from inside add. Collectors exist for this: each worker
 *   accumulates into its own container and the results are merged, which is
 *   both correct and faster than contending on one list.
 *
 *   Defect 3 — pairwise averaging. It is not associative, so in parallel the
 *   answer depends on how the stream was split and therefore on the core
 *   count. It is also simply wrong sequentially, weighting later elements far
 *   more heavily. average() is correct, and its parallel form is correct too
 *   because summing IS associative.
 */
public class Solution {

    record Charge(String id, double amount, boolean settled) { }

    static final List<String> IDS =
        IntStream.range(0, 40).mapToObj(i -> "C-" + i).toList();

    static final Set<String> fetchThreads = ConcurrentHashMap.newKeySet();

    static Charge fetch(String id) {
        fetchThreads.add(Thread.currentThread().getName());
        try {
            Thread.sleep(5);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        int n = Integer.parseInt(id.substring(2));
        return new Charge(id, 10.0 + n, n % 3 != 0);
    }

    /** FIX 1: a pool this job owns, sized for the remote service. */
    static List<Charge> fetchAll(List<String> ids) throws Exception {
        ExecutorService io = Executors.newFixedThreadPool(8);
        try {
            List<Future<Charge>> futures = ids.stream()
                .map(id -> io.submit(() -> fetch(id)))
                .toList();
            List<Charge> out = new ArrayList<>(futures.size());
            for (Future<Charge> f : futures) out.add(f.get());
            return List.copyOf(out);
        } finally {
            io.shutdown();
        }
    }

    /** FIX 2: a collector, so each worker accumulates separately. */
    static List<Charge> settledOnly(List<Charge> charges) {
        return charges.parallelStream().filter(Charge::settled).toList();
    }

    /** FIX 3: summing is associative, so this is correct in parallel too. */
    static double averageAmount(List<Charge> charges) {
        return charges.parallelStream().mapToDouble(Charge::amount).average().orElseThrow();
    }

    public static void main(String[] args) throws Exception {
        boolean ok = true;

        List<Charge> all = fetchAll(IDS);
        List<Charge> settled = settledOnly(all);
        long expectedSettled = all.stream().filter(Charge::settled).count();
        double expectedAverage = all.stream().mapToDouble(Charge::amount).average().orElseThrow();
        double average = averageAmount(all);

        System.out.println("── the nightly report ──");
        System.out.println("  fetched        : " + all.size() + " of " + IDS.size());
        System.out.println("  fetch threads  : " + describe(fetchThreads));
        System.out.println("  settled found  : " + settled.size() + " of " + expectedSettled);
        System.out.printf ("  average        : %.4f (correct: %.4f)%n", average, expectedAverage);
        System.out.println();

        ok &= check("no blocking call ran on the shared common pool",
            fetchThreads.stream().noneMatch(n -> n.startsWith("ForkJoinPool.commonPool")));
        ok &= check("every settled charge was collected", settled.size() == expectedSettled);
        ok &= check("the average is correct",
            Math.abs(average - expectedAverage) < 0.0001);

        System.out.println();
        System.out.println(ok ? "PASS" : "FAIL");
    }

    static String describe(Set<String> names) {
        boolean common = names.stream().anyMatch(n -> n.startsWith("ForkJoinPool.commonPool"));
        boolean caller = names.contains("main");
        boolean own = names.stream().anyMatch(n -> n.startsWith("pool-"));
        return (common ? "common pool" : own ? "our own pool" : "unknown")
            + (caller ? " + the calling thread" : "");
    }

    static boolean check(String what, boolean passed) {
        System.out.println((passed ? "  ok    " : "  FAIL  ") + what);
        return passed;
    }
}