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

public class CounterLab {
    interface Counter { void increment(); long get(); }

    static final class Unsafe implements Counter { private long n; public void increment() { n++; } public long get() { return n; } }
    static final class Locked implements Counter { private long n; public synchronized void increment() { n++; } public synchronized long get() { return n; } }
    static final class Atomic implements Counter { private final AtomicLong n = new AtomicLong(); public void increment() { n.incrementAndGet(); } public long get() { return n.get(); } }
    static final class Adder implements Counter { private final LongAdder n = new LongAdder(); public void increment() { n.increment(); } public long get() { return n.sum(); } }

    static long[] run(Counter c, int threads, int perThread) throws InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(threads);
        CountDownLatch start = new CountDownLatch(1), done = new CountDownLatch(threads);
        for (int t = 0; t < threads; t++) {
            pool.execute(() -> {
                try {
                    start.await();
                    for (int i = 0; i < perThread; i++) c.increment();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    done.countDown();
                }
            });
        }
        long t0 = System.nanoTime();
        start.countDown();
        done.await();
        long ms = (System.nanoTime() - t0) / 1_000_000;
        pool.shutdown();
        return new long[] { c.get(), ms };      // read after done.await(): happens-before via the latch
    }

    public static void main(String[] args) throws InterruptedException {
        int threads = 8, perThread = 1_000_000;
        long expected = (long) threads * perThread;
        for (Counter c : new Counter[] { new Unsafe(), new Locked(), new Atomic(), new Adder() }) {
            long[] r = run(c, threads, perThread);
            System.out.printf("%-7s value=%,d correct=%b time=%d ms%n", c.getClass().getSimpleName(), r[0], r[0] == expected, r[1]);
        }
    }
}

Unsafe  value=2,915,734 correct=false time=21 ms
Locked  value=8,000,000 correct=true time=412 ms
Atomic  value=8,000,000 correct=true time=158 ms
Adder   value=8,000,000 correct=true time=37 ms

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

public class BufferLab {
    interface Buffer<T> { void put(T t) throws InterruptedException; T take() throws InterruptedException; int size(); }

    static final class MonitorBuffer<T> implements Buffer<T> {
        private final Deque<T> q = new ArrayDeque<>(); private final int cap;
        MonitorBuffer(int cap) { this.cap = cap; }
        public synchronized void put(T t) throws InterruptedException { while (q.size() == cap) wait(); q.addLast(t); notifyAll(); }
        public synchronized T take() throws InterruptedException { while (q.isEmpty()) wait(); T t = q.removeFirst(); notifyAll(); return t; }
        public synchronized int size() { return q.size(); }
    }

    static final class ConditionBuffer<T> implements Buffer<T> {
        private final Deque<T> q = new ArrayDeque<>(); private final int cap;
        private final ReentrantLock lock = new ReentrantLock();
        private final Condition notFull = lock.newCondition(), notEmpty = lock.newCondition();
        ConditionBuffer(int cap) { this.cap = cap; }
        public void put(T t) throws InterruptedException {
            lock.lock();
            try { while (q.size() == cap) notFull.await(); q.addLast(t); notEmpty.signal(); }
            finally { lock.unlock(); }
        }
        public T take() throws InterruptedException {
            lock.lock();
            try { while (q.isEmpty()) notEmpty.await(); T t = q.removeFirst(); notFull.signal(); return t; }
            finally { lock.unlock(); }
        }
        public int size() { lock.lock(); try { return q.size(); } finally { lock.unlock(); } }
    }

    static final class QueueBuffer<T> implements Buffer<T> {
        private final BlockingQueue<T> q;
        QueueBuffer(int cap) { q = new ArrayBlockingQueue<>(cap); }
        public void put(T t) throws InterruptedException { q.put(t); }
        public T take() throws InterruptedException { return q.take(); }
        public int size() { return q.size(); }
    }

    static void test(String name, Buffer<Integer> buf, int cap) throws Exception {
        int producers = 4, consumers = 4, perProducer = 25_000;
        AtomicLong sum = new AtomicLong();
        AtomicInteger maxSize = new AtomicInteger();
        ExecutorService pool = Executors.newFixedThreadPool(producers + consumers);
        List<Future<?>> fs = new ArrayList<>();
        for (int p = 0; p < producers; p++)
            fs.add(pool.submit(() -> { for (int i = 1; i <= perProducer; i++) { buf.put(i); maxSize.accumulateAndGet(buf.size(), Math::max); } return null; }));
        for (int c = 0; c < consumers; c++)
            fs.add(pool.submit(() -> { for (int i = 0; i < perProducer; i++) sum.addAndGet(buf.take()); return null; }));
        try {
            for (Future<?> f : fs) f.get(30, TimeUnit.SECONDS); // a lost wakeup shows up as a TimeoutException
        } finally {
            pool.shutdownNow();                                  // interrupt stuck waiters so the JVM can exit
        }
        long expected = (long) producers * perProducer * (perProducer + 1L) / 2;
        System.out.printf("%-18s sum ok=%b, max size seen=%d (capacity %d)%n", name, sum.get() == expected, maxSize.get(), cap);
    }

    public static void main(String[] args) throws Exception {
        test("wait/notifyAll", new MonitorBuffer<>(16), 16);
        test("Condition", new ConditionBuffer<>(16), 16);
        test("ArrayBlockingQueue", new QueueBuffer<>(16), 16);
    }
}

import java.util.concurrent.Semaphore;

public class TakeTurns {
    public static void main(String[] args) throws InterruptedException {
        int max = 10;
        Semaphore oddTurn = new Semaphore(1), evenTurn = new Semaphore(0);
        StringBuilder out = new StringBuilder();   // safe: every access is ordered by release -> acquire

        Thread odd = new Thread(() -> {
            try {
                for (int i = 1; i <= max; i += 2) { oddTurn.acquire(); out.append(i).append(' '); evenTurn.release(); }
            } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        }, "odd");
        Thread even = new Thread(() -> {
            try {
                for (int i = 2; i <= max; i += 2) { evenTurn.acquire(); out.append(i).append(' '); oddTurn.release(); }
            } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        }, "even");

        odd.start(); even.start();
        odd.join(); even.join();                   // join: main sees the final contents
        System.out.println(out.toString().trim());
    }
}

public class TakeTurnsMonitor {
    private final int max;
    private int next = 1;                          // guarded by this
    private final StringBuilder out = new StringBuilder();   // guarded by this

    TakeTurnsMonitor(int max) { this.max = max; }

    synchronized void loop(boolean oddThread) throws InterruptedException {
        while (true) {
            while (next <= max && (next % 2 == 1) != oddThread) wait();   // not my turn
            if (next > max) { notifyAll(); return; }                      // done: wake the other so it can exit too
            out.append(next).append(' ');
            next++;
            notifyAll();                                                  // pass the turn
        }
    }

    public static void main(String[] args) throws InterruptedException {
        TakeTurnsMonitor m = new TakeTurnsMonitor(10);
        Thread odd = new Thread(() -> { try { m.loop(true); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } });
        Thread even = new Thread(() -> { try { m.loop(false); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } });
        odd.start(); even.start();
        odd.join(); even.join();
        synchronized (m) { System.out.println(m.out.toString().trim()); }
    }
}

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

public class Memoizer<K, V> {
    private final ConcurrentHashMap<K, CompletableFuture<V>> cache = new ConcurrentHashMap<>();
    private final Function<K, V> loader;
    private final Executor executor;

    public Memoizer(Function<K, V> loader, Executor executor) { this.loader = loader; this.executor = executor; }

    public CompletableFuture<V> get(K key) {
        // Atomic and fast: the bin is locked only while the future object is created, not during the load.
        CompletableFuture<V> f = cache.computeIfAbsent(key, k -> CompletableFuture.supplyAsync(() -> loader.apply(k), executor));
        f.whenComplete((v, ex) -> { if (ex != null) cache.remove(key, f); });   // never cache a failure
        return f.copy();       // callers get a copy: their orTimeout or cancel cannot complete the cached future
    }

    public static void main(String[] args) throws Exception {
        AtomicInteger loads = new AtomicInteger();
        ExecutorService io = Executors.newFixedThreadPool(4);
        Memoizer<String, String> rates = new Memoizer<>(pair -> {
            loads.incrementAndGet();
            try { Thread.sleep(200); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            return pair + "=83.12";
        }, io);

        ExecutorService callers = Executors.newFixedThreadPool(10);
        CountDownLatch start = new CountDownLatch(1);
        List<Future<String>> results = new ArrayList<>();
        for (int i = 0; i < 10; i++)
            results.add(callers.submit(() -> { start.await(); return rates.get("USD/INR").get(1, TimeUnit.SECONDS); }));
        start.countDown();
        for (Future<String> f : results) f.get();
        System.out.println("10 callers, loads = " + loads.get() + ", value = " + results.get(0).get());
        callers.shutdown();
        io.shutdown();
    }
}

import java.util.*;
import java.util.concurrent.*;

public class Aggregator {
    record Quote(String partner, long pricePaise) { }

    static Quote callPartner(String name, long latencyMs, boolean fail) {
        try { Thread.sleep(latencyMs); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        if (fail) throw new IllegalStateException(name + " returned 500");
        return new Quote(name, 10_000 + latencyMs);
    }

    static CompletableFuture<Optional<Quote>> quote(Executor io, String name, long ms, boolean fail) {
        return CompletableFuture.supplyAsync(() -> callPartner(name, ms, fail), io)
                .thenApply(Optional::of)
                .completeOnTimeout(Optional.empty(), 300, TimeUnit.MILLISECONDS)   // too slow: ignore
                .exceptionally(ex -> Optional.empty());                            // failed: ignore
    }

    public static void main(String[] args) {
        ExecutorService io = new ThreadPoolExecutor(8, 8, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100),
                Thread.ofPlatform().name("partner-io-", 0).factory());   // bounded: a full queue rejects at supplyAsync
        long start = System.nanoTime();
        List<CompletableFuture<Optional<Quote>>> calls = List.of(
                quote(io, "alpha", 120, false),
                quote(io, "beta", 900, false),
                quote(io, "gamma", 80, true),
                quote(io, "delta", 200, false));

        List<Quote> got = CompletableFuture.allOf(calls.toArray(new CompletableFuture[0]))
                .thenApply(v -> calls.stream().map(CompletableFuture::join).flatMap(Optional::stream).toList())
                .join();
        Quote best = got.stream().min(Comparator.comparingLong(Quote::pricePaise)).orElseThrow();
        long ms = (System.nanoTime() - start) / 1_000_000;

        System.out.println("quotes from " + got.stream().map(Quote::partner).toList() + ", best " + best.partner() + ", in " + ms + " ms");
        io.shutdownNow();     // interrupts beta's abandoned call
    }
}

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

public class OverloadLab {
    static void burst(String label, RejectedExecutionHandler policy) throws InterruptedException {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 4, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10),
                Thread.ofPlatform().name("job-", 0).factory(), policy);
        AtomicInteger done = new AtomicInteger(), rejected = new AtomicInteger(), onCaller = new AtomicInteger();
        String submitter = Thread.currentThread().getName();

        long start = System.nanoTime();
        for (int i = 0; i < 30; i++) {
            try {
                pool.execute(() -> {
                    if (Thread.currentThread().getName().equals(submitter)) onCaller.incrementAndGet();
                    try { Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
                    done.incrementAndGet();
                });
            } catch (RejectedExecutionException e) {
                rejected.incrementAndGet();          // in a service: return 429/503 here
            }
        }
        long submitMs = (System.nanoTime() - start) / 1_000_000;

        pool.shutdown();
        if (!pool.awaitTermination(10, TimeUnit.SECONDS)) pool.shutdownNow();
        System.out.printf("%-10s completed=%d rejected=%d ranOnCaller=%d largestPool=%d submitLoop=%d ms%n",
                label, done.get(), rejected.get(), onCaller.get(), pool.getLargestPoolSize(), submitMs);
    }

    public static void main(String[] args) throws InterruptedException {
        burst("Abort", new ThreadPoolExecutor.AbortPolicy());
        burst("CallerRuns", new ThreadPoolExecutor.CallerRunsPolicy());
    }
}

Abort      completed=14 rejected=16 ranOnCaller=0 largestPool=4 submitLoop=3 ms
CallerRuns completed=30 rejected=0 ranOnCaller=4 largestPool=4 submitLoop=402 ms

import java.lang.management.ManagementFactory;
import java.util.concurrent.*;

public class DeadlockLab {
    static final class Account { final int id; long balance; Account(int id, long b) { this.id = id; this.balance = b; } }
    interface Transfer { void run(Account from, Account to, long amount); }

    static void naive(Account from, Account to, long amt) {
        synchronized (from) { synchronized (to) { from.balance -= amt; to.balance += amt; } }
    }
    static void ordered(Account from, Account to, long amt) {
        Account first = from.id < to.id ? from : to, second = (first == from) ? to : from;   // global order: by id
        synchronized (first) { synchronized (second) { from.balance -= amt; to.balance += amt; } }
    }

    static void simulate(String label, Transfer transfer) throws InterruptedException {
        Account[] accts = new Account[5];
        for (int i = 0; i < accts.length; i++) accts[i] = new Account(i, 1_000);
        CountDownLatch done = new CountDownLatch(8);
        for (int t = 0; t < 8; t++) {
            Thread th = new Thread(() -> {
                ThreadLocalRandom r = ThreadLocalRandom.current();
                for (int i = 0; i < 100_000; i++) {
                    int a = r.nextInt(accts.length), b = r.nextInt(accts.length);
                    if (a != b) transfer.run(accts[a], accts[b], 1);
                }
                done.countDown();
            }, label + "-" + t);
            th.setDaemon(true);                          // stuck demo threads must not keep the JVM alive
            th.start();
        }
        if (done.await(5, TimeUnit.SECONDS)) {
            long total = 0;
            for (Account a : accts) synchronized (a) { total += a.balance; }   // safe: all transfers finished (latch)
            System.out.println(label + ": all transfers finished, total = " + total);
        } else {
            long[] ids = ManagementFactory.getThreadMXBean().findDeadlockedThreads();
            System.out.println(label + ": stuck, deadlocked threads = " + (ids == null ? 0 : ids.length));
        }
    }

    public static void main(String[] args) throws InterruptedException {
        simulate("naive", DeadlockLab::naive);
        simulate("ordered", DeadlockLab::ordered);
    }
}

naive: stuck, deadlocked threads = 2
ordered: all transfers finished, total = 5000

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.function.Supplier;

public class BulkheadLab {
    static final class Bulkhead {
        private final String name; private final Semaphore permits; private final long waitMs;
        final LongAdder rejected = new LongAdder();
        final AtomicInteger maxInFlight = new AtomicInteger();
        private final AtomicInteger inFlight = new AtomicInteger();

        Bulkhead(String name, int limit, long waitMs) { this.name = name; this.permits = new Semaphore(limit); this.waitMs = waitMs; }

        <T> T call(Supplier<T> work) {
            boolean ok;
            try { ok = permits.tryAcquire(waitMs, TimeUnit.MILLISECONDS); }
            catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IllegalStateException("interrupted"); }
            if (!ok) { rejected.increment(); throw new RejectedExecutionException(name + " bulkhead full"); }
            try {
                maxInFlight.accumulateAndGet(inFlight.incrementAndGet(), Math::max);
                return work.get();
            } finally {
                inFlight.decrementAndGet();
                permits.release();
            }
        }
    }

    static void sleep(long ms) { try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }

    static long run(Bulkhead notifications) throws Exception {
        ExecutorService requests = Executors.newFixedThreadPool(40);      // the shared request threads
        LongAdder paymentMs = new LongAdder(); LongAdder payments = new LongAdder();
        List<Future<?>> fs = new ArrayList<>();
        for (int i = 0; i < 200; i++) {
            long submitted = System.nanoTime();
            boolean payment = i % 2 == 0;
            fs.add(requests.submit(() -> {
                if (payment) {
                    sleep(20);                                                // fraud check
                    paymentMs.add((System.nanoTime() - submitted) / 1_000_000);
                    payments.increment();
                } else {
                    try { notifications.call(() -> { sleep(1_000); return "sent"; }); }
                    catch (RejectedExecutionException e) { /* degrade: store the receipt and send it later */ }
                }
            }));
        }
        for (Future<?> f : fs) f.get(60, TimeUnit.SECONDS);
        requests.shutdown();
        return paymentMs.sum() / payments.sum();
    }

    public static void main(String[] args) throws Exception {
        long without = run(new Bulkhead("no limit", 1_000, 50));
        Bulkhead bh = new Bulkhead("notifications", 5, 50);
        long with = run(bh);
        System.out.println("average payment latency without bulkhead: " + without + " ms");
        System.out.println("average payment latency with bulkhead:    " + with + " ms"
                + " (notifications max in flight " + bh.maxInFlight.get() + ", rejected " + bh.rejected.sum() + ")");
        System.out.println("bulkhead kept payments fast: " + (with * 2 < without));
    }
}

average payment latency without bulkhead: 1186 ms
average payment latency with bulkhead:    41 ms (notifications max in flight 5, rejected 95)
bulkhead kept payments fast: true

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.function.LongSupplier;

public class RateLimiterLab {
    static final class TokenBucket {
        private final long capacity, nanosPerToken;
        private final LongSupplier clock;
        private long tokens, lastRefill;                  // guarded by this

        TokenBucket(long capacity, long tokensPerSecond, LongSupplier clock) {
            this.capacity = capacity;
            this.nanosPerToken = 1_000_000_000L / tokensPerSecond;
            this.clock = clock;
            this.tokens = capacity;                       // start full: allows one burst
            this.lastRefill = clock.getAsLong();
        }

        synchronized boolean tryAcquire() {               // refill and take are one atomic step
            refill();
            if (tokens == 0) return false;
            tokens--;
            return true;
        }

        private void refill() {                           // lazy: no background thread
            long now = clock.getAsLong();
            long earned = (now - lastRefill) / nanosPerToken;
            tokens = Math.min(capacity, tokens + earned);
            // a full bucket restarts the clock; otherwise keep the unused part of a token
            lastRefill = (tokens == capacity) ? now : lastRefill + earned * nanosPerToken;
        }
    }

    // Many threads call the limiter at the same moment; returns how many calls got a token
    static int hammer(TokenBucket b, int threads, int callsEach) throws Exception {
        AtomicInteger allowed = new AtomicInteger();
        CountDownLatch gate = new CountDownLatch(1);
        ExecutorService pool = Executors.newFixedThreadPool(threads);
        List<Future<?>> fs = new ArrayList<>();
        for (int t = 0; t < threads; t++) {
            fs.add(pool.submit(() -> {
                gate.await();
                for (int i = 0; i < callsEach; i++) if (b.tryAcquire()) allowed.incrementAndGet();
                return null;
            }));
        }
        gate.countDown();
        for (Future<?> f : fs) f.get(10, TimeUnit.SECONDS);
        pool.shutdown();
        return allowed.get();
    }

    public static void main(String[] args) throws Exception {
        AtomicLong fakeNanos = new AtomicLong();                        // a clock the test moves by hand
        TokenBucket b = new TokenBucket(10, 100, fakeNanos::get);       // bursts of 10, 100 per second

        System.out.println("burst: " + hammer(b, 8, 100) + " allowed");                     // 800 calls at once
        fakeNanos.addAndGet(50_000_000);                                                     // 50 ms later
        System.out.println("after 50 ms: " + hammer(b, 8, 100) + " allowed");
        fakeNanos.addAndGet(1_000_000_000);                                                  // 1 s idle
        System.out.println("after 1 s idle: " + hammer(b, 8, 100) + " allowed (capped at capacity)");

        // One bucket per merchant, created on first use
        ConcurrentHashMap<String, TokenBucket> perMerchant = new ConcurrentHashMap<>();
        int noisy = 0, quiet = 0;
        for (int i = 0; i < 50; i++)
            if (perMerchant.computeIfAbsent("noisy", k -> new TokenBucket(5, 10, System::nanoTime)).tryAcquire()) noisy++;
        for (int i = 0; i < 3; i++)
            if (perMerchant.computeIfAbsent("quiet", k -> new TokenBucket(5, 10, System::nanoTime)).tryAcquire()) quiet++;
        System.out.println("noisy merchant: " + noisy + " of 50 allowed, quiet merchant: " + quiet + " of 3 allowed");
    }
}

burst: 10 allowed
after 50 ms: 5 allowed
after 1 s idle: 10 allowed (capped at capacity)
noisy merchant: 5 of 50 allowed, quiet merchant: 3 of 3 allowed

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

public class JobRunner implements AutoCloseable {
    enum Status {
        QUEUED, RUNNING, RETRY_WAIT, SUCCEEDED, FAILED, REJECTED, ABANDONED;
        boolean isFinal() { return this == SUCCEEDED || this == FAILED || this == REJECTED || this == ABANDONED; }
    }
    record Job(String id, Callable<String> work) { }

    final ConcurrentHashMap<String, Status> status = new ConcurrentHashMap<>();
    final LongAdder accepted = new LongAdder(), rejected = new LongAdder(), succeeded = new LongAdder(),
            failed = new LongAdder(), abandoned = new LongAdder(), retries = new LongAdder(), timeouts = new LongAdder();

    private final ThreadPoolExecutor workers;
    private final ScheduledThreadPoolExecutor timer;                // watchdogs and retry delays
    private final int maxAttempts;
    private final long timeoutMs, backoffMs;
    private volatile boolean closing;

    JobRunner(int threads, int queueSize, int maxAttempts, long timeoutMs, long backoffMs) {
        workers = new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(queueSize), Thread.ofPlatform().name("job-worker-", 0).factory());
        timer = new ScheduledThreadPoolExecutor(1, Thread.ofPlatform().name("job-timer").daemon(true).factory());
        timer.setRemoveOnCancelPolicy(true);                       // cancelled watchdogs leave the queue at once
        this.maxAttempts = maxAttempts; this.timeoutMs = timeoutMs; this.backoffMs = backoffMs;
    }

    /** False if the runner is closing, the id was already accepted, or the queue is full (a rejected id may try again). */
    public boolean submit(String id, Callable<String> work) {
        if (closing) return false;
        if (status.putIfAbsent(id, Status.QUEUED) != null
                && !status.replace(id, Status.REJECTED, Status.QUEUED)) return false;   // each id runs once
        boolean ok = start(new Job(id, work), 1);
        if (ok) accepted.increment();
        return ok;
    }

    private boolean start(Job job, int attempt) {
        try {
            workers.execute(new Attempt(job, attempt));
            return true;
        } catch (RejectedExecutionException e) {                   // queue full, or shut down
            end(job, closing ? Status.ABANDONED : attempt == 1 ? Status.REJECTED : Status.FAILED);   // a retry that finds the queue full gives up
            return false;
        }
    }

    /** One attempt of one job. done() runs exactly once: after a return, a throw or a cancel. */
    private final class Attempt extends FutureTask<String> {
        private final Job job; private final int n;
        Attempt(Job job, int n) { super(job.work()); this.job = job; this.n = n; }

        @Override public void run() {                                // on a worker thread
            status.put(job.id(), Status.RUNNING);
            ScheduledFuture<?> watchdog = timer.schedule(() -> cancel(true), timeoutMs, TimeUnit.MILLISECONDS);
            try { super.run(); } finally { watchdog.cancel(false); }
        }

        @Override protected void done() { finished(job, n, this); }  // worker thread or timer thread
    }

    private void finished(Job job, int n, Future<String> attempt) {
        Future.State s = attempt.state();                            // Java 19+: SUCCESS, FAILED or CANCELLED
        if (s == Future.State.SUCCESS) { end(job, Status.SUCCEEDED); return; }
        if (s == Future.State.CANCELLED) timeouts.increment();       // only the watchdog cancels attempts
        if (n == maxAttempts) { end(job, Status.FAILED); return; }
        if (closing) { end(job, Status.ABANDONED); return; }         // no new retries during shutdown
        status.put(job.id(), Status.RETRY_WAIT);
        retries.increment();
        long delay = backoffMs << (n - 1);                           // 50, 100, 200 ms ...
        delay += ThreadLocalRandom.current().nextLong(delay / 2 + 1); // jitter: retries do not collide
        timer.schedule(() -> { status.put(job.id(), Status.QUEUED); start(job, n + 1); }, delay, TimeUnit.MILLISECONDS);
    }

    private void end(Job job, Status s) {
        Status old = status.get(job.id());
        if (old.isFinal() || !status.replace(job.id(), old, s)) return;   // the first final state wins
        switch (s) {
            case SUCCEEDED -> succeeded.increment();
            case FAILED -> failed.increment();
            case REJECTED -> rejected.increment();
            default -> abandoned.increment();
        }
    }

    @Override public void close() throws InterruptedException {
        closing = true;                                              // 1. refuse new jobs and new retries
        workers.shutdown();                                          // 2. queued and running attempts may finish
        if (!workers.awaitTermination(5, TimeUnit.SECONDS)) {
            workers.shutdownNow();                                   //    grace period over: interrupt them
            workers.awaitTermination(1, TimeUnit.SECONDS);
        }
        timer.shutdownNow();                                         // 3. drop pending retries
        timer.awaitTermination(1, TimeUnit.SECONDS);
        status.forEach((id, s) -> {                                  // 4. nothing stays half done
            if (!s.isFinal() && status.replace(id, s, Status.ABANDONED)) abandoned.increment();
        });
    }

    String metrics() {
        return "accepted=" + accepted + ", rejected=" + rejected + ", succeeded=" + succeeded + ", failed=" + failed
                + ", abandoned=" + abandoned + ", retries=" + retries + ", timeouts=" + timeouts;
    }

    public static void main(String[] args) throws Exception {
        JobRunner runner = new JobRunner(4, 20, 3, 200, 50);   // 4 workers, queue 20, 3 attempts, 200 ms timeout, 50 ms backoff
        AtomicInteger flakyCalls = new AtomicInteger();
        runner.submit("flaky", () -> {
            Thread.sleep(20);
            if (flakyCalls.incrementAndGet() < 3) throw new IllegalStateException("partner returned 503");
            return "ok";
        });
        runner.submit("broken", () -> { Thread.sleep(20); throw new IllegalArgumentException("bad input"); });
        runner.submit("slow", () -> { Thread.sleep(5_000); return "too late"; });     // the watchdog cancels every attempt

        int ok = 0;
        for (int i = 1; i <= 40; i++) if (runner.submit("job-" + i, () -> { Thread.sleep(30); return "done"; })) ok++;
        System.out.println("burst of 40: " + ok + " accepted, " + (40 - ok) + " rejected");

        Thread.sleep(2_000);                                     // let retries and timeouts play out
        System.out.println("flaky: " + runner.status.get("flaky") + " after " + flakyCalls.get() + " attempts");
        System.out.println("broken: " + runner.status.get("broken"));
        System.out.println("slow: " + runner.status.get("slow"));

        for (int i = 1; i <= 8; i++) runner.submit("late-" + i, () -> { Thread.sleep(100); return "done"; });
        runner.close();                                          // graceful: the 8 late jobs still finish
        long drained = IntStream.rangeClosed(1, 8).filter(i -> runner.status.get("late-" + i) == Status.SUCCEEDED).count();
        System.out.println("drained during shutdown: " + drained + " of 8 SUCCEEDED");
        System.out.println("submit after close accepted: " + runner.submit("too-late", () -> "x"));
        System.out.println(runner.metrics());
        System.out.println("every job has a final status: " + runner.status.values().stream().allMatch(Status::isFinal));
    }
}