The queue that ate the heap

import java.util.concurrent.*;

/**
 * Solution: the queue that ate the heap.
 *
 * All three symptoms came from one line: the queue.
 *
 * A ThreadPoolExecutor uses its settings in an order most people guess wrong.
 * Core threads first, then the QUEUE, and only once the queue is full does it
 * create threads up to maximumPoolSize. Rejection is the step after that.
 *
 * `new LinkedBlockingQueue<>()` with no capacity argument holds
 * Integer.MAX_VALUE tasks, so the queue never filled. Which means:
 *
 *   the pool never grew past core        — step 3 was unreachable
 *   nothing was ever rejected            — step 4 was unreachable
 *   every upload was accepted and kept   — payload retained on the heap
 *
 * maximumPoolSize and the AbortPolicy were not misconfigured. They were
 * unreachable, which is worse, because both read as protection in review.
 *
 * THE ANSWER TO OPS
 *   maximumPoolSize=100 would have changed nothing at all. The pool was
 *   stuck at 2 because the queue never filled, and raising a limit that is
 *   never consulted is a no-op. Bounding the queue is what makes every other
 *   dial mean something.
 *
 * THE HONEST COST
 *   The service now visibly refuses uploads it previously accepted. That is
 *   not a regression: it previously accepted them and then failed to deliver
 *   them in time, holding each payload in memory while the client timed out.
 *   A refusal at submit time is a 503 the caller can retry; a queued task is
 *   latency already promised to someone who has usually stopped waiting.
 *
 *   The number to alert on is queue depth, not thread count. Thread count is
 *   a lagging indicator that, with an unbounded queue, never moves at all.
 */
public class Solution {

    static final int TASKS = 200;
    static final int CORE = 2;
    static final int MAX = 20;

    /** What the pool is allowed to owe before we consider it overloaded. */
    static final int ACCEPTABLE_QUEUE_DEPTH = 25;

    public static void main(String[] args) throws Exception {
        /*
         * FIX: bound the queue. Everything else was already correct and had
         * simply never been reached.
         *
         * The capacity is a real decision, not a default: it is how much
         * latency the service is willing to promise. 20 slots against 20
         * threads is roughly "one queued task per worker", which keeps the
         * debt visible and small.
         */
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                CORE, MAX,
                60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(20),
                new ThreadPoolExecutor.AbortPolicy());

        CountDownLatch release = new CountDownLatch(1);
        int refused = 0;

        for (int i = 0; i < TASKS; i++) {
            try {
                pool.execute(() -> {
                    try {
                        release.await();          // stands in for slow I/O
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                });
            } catch (RejectedExecutionException e) {
                refused++;
            }
        }

        Thread.sleep(200);

        int threads = pool.getLargestPoolSize();
        int queued = pool.getQueue().size();

        System.out.println("── " + TASKS + " uploads, pool core=" + CORE + " max=" + MAX + " ──");
        System.out.println("  threads the pool actually made : " + threads + "   (configured for " + MAX + ")");
        System.out.println("  tasks waiting in the queue     : " + queued);
        System.out.println("  uploads refused                : " + refused);

        release.countDown();
        pool.shutdown();
        pool.awaitTermination(10, TimeUnit.SECONDS);

        boolean usedItsThreads = threads == MAX;
        boolean queueStayedBounded = queued <= ACCEPTABLE_QUEUE_DEPTH;
        boolean overloadWasSignalled = refused > 0;

        System.out.println();
        System.out.println("pool used the threads it was given : " + usedItsThreads);
        System.out.println("queue depth stayed bounded         : " + queueStayedBounded);
        System.out.println("overload was signalled, not hidden : " + overloadWasSignalled);
        System.out.println(
                usedItsThreads && queueStayedBounded && overloadWasSignalled ? "PASS" : "FAIL");
    }
}