ThreadPoolExecutor: Stop Using Factory Methods in Production

Revolut Interview Prep — Day 5 of 16

It’s Monday morning. Revolut’s payment service is handling the post-weekend backlog — 50,000 tasks queued up. The system uses Executors.newFixedThreadPool(10). Memory climbs. Climbs. Then the OOM killer fires. The pod restarts. Payments are lost.

The culprit? One line of code. A line that looks completely reasonable.

Day 4 showed us how to avoid locks with CAS. Today we look at what happens when you do need threads — and why the convenience wrappers will get you paged at 3am.

Why thread pools exist

Creating a thread is expensive — 512KB to 1MB of stack, OS registration, scheduler overhead. Do it per-request and you pay that cost on every payment. Thread pools solve three problems at once: reuse threads instead of creating them, throttle parallelism to match your hardware, and apply back pressure when the system is overwhelmed.

The Executor framework hierarchy is straightforward: Executor → ExecutorService → AbstractExecutorService → ThreadPoolExecutor→ ScheduledThreadPoolExecutor. Everything you care about lives in ThreadPoolExecutor.

The constructor you need to know

new ThreadPoolExecutor(
    int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler
)

Seven parameters. Every single one matters in production. Let’s go through them.

How the pool actually grows (the interview trap)

Most people get this wrong. The growth algorithm is not intuitive:

Task arrives →
  if (active threads < corePoolSize)  → create new thread (even if idle threads exist!)
  else if (queue is not full)         → put in queue
  else if (active threads < maxPool)  → create new thread
  else                                → call RejectedExecutionHandler

The trap: below corePoolSize, a new thread is always created — even if idle threads are waiting. The pool fills to core first, then queues, then grows to max only after the queue is full.

This means if your queue is unbounded, maximumPoolSize is never reached. You have one knob that does nothing.

The parameters in detail

**corePoolSize** — threads that stay alive always, even when idle.

// I/O bound (database, external APIs)
int core = Runtime.getRuntime().availableProcessors() * 2;
// CPU bound (validation, encryption)
int core = Runtime.getRuntime().availableProcessors() + 1;

**maximumPoolSize** — the ceiling. Only relevant after the queue fills up. If your queue is unbounded, this parameter is dead code.

**keepAliveTime** — how long a thread above corePoolSize stays alive when idle. For instant payments, keep it short (10 seconds) — release resources fast. For batch jobs, keep it longer (120 seconds) — recreating threads is expensive.

**threadFactory** — non-negotiable in production:

@Override
public Thread newThread(Runnable r) {
    Thread t = new Thread(r);
    t.setName("revolut-" + poolName + "-worker-" + count.getAndIncrement());
    t.setDaemon(false); // daemon dies with main thread — payments won't finish
    t.setPriority(Thread.NORM_PRIORITY);
    t.setUncaughtExceptionHandler((thread, e) -> {
        log.error("Uncaught exception in thread {}", thread.getName(), e);
    });
    return t;
}

daemon = false is critical. Daemon threads are killed the moment the main thread exits — your in-flight payments never complete. Named threads are the difference between a readable thread dump and an hour of debugging pool-1-thread-47.

The queues

**SynchronousQueue** — stores nothing. Every submitted task requires a free thread right now. If none exists and max is reached, it rejects immediately. Used internally by Executors.newCachedThreadPool() — with Integer.MAX_VALUE max threads, meaning threads grow to OOM under burst load.

// What newCachedThreadPool() actually is:
new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, SECONDS, new SynchronousQueue<>());
//                        ^^^^^^^^^^^^^^^^^
//                        2 billion threads → OOM

Good for: short-lived tasks with unpredictable burst, push notifications. Dangerous without a bounded max.

**LinkedBlockingQueue** — two separate locks for put and take, high throughput. But the default constructor is unbounded:

new LinkedBlockingQueue<>()        // unbounded — OOM waiting to happen
new LinkedBlockingQueue<>(10_000)  // bounded — correct

This is exactly what Executors.newFixedThreadPool() uses under the hood — unbounded. Queue grows forever, memory fills up, pod dies.

**ArrayBlockingQueue** — memory allocated upfront, one lock shared between put and take. Lower throughput than LinkedBlockingQueue, but fully predictable memory footprint:

new ArrayBlockingQueue<>(1000)        // memory pre-allocated
new ArrayBlockingQueue<>(1000, true)  // fair mode — strict FIFO

ArrayBlockingQueue pre-allocates all its memory at creation. LinkedBlockingQueue allocates nodes as tasks arrive. The tradeoff: ArrayBlockingQueue is predictable but uses memory upfront; LinkedBlockingQueue is more memory-efficient at low load but has higher throughput under contention due to its split-lock design.

For payment processing, ArrayBlockingQueue wins — predictable size, controlled memory, no surprises.

⚠️ Queue size must match your SLA. If SLA is 2 seconds and processing takes 500ms, a queue of 1000 tasks means 500 seconds of wait time. That’s not back pressure — that’s a hidden outage. Keep the queue small enough that rejection happens before SLA is violated.

Rejection policies

Rejection triggers when the queue is full and maximumPoolSize is reached. What happens next depends on your RejectedExecutionHandler.

**AbortPolicy (default)** — throws RejectedExecutionException. The caller catches it and returns HTTP 429. Honest, fast, appropriate for REST endpoints and payment APIs.

**CallerRunsPolicy** — the submitted task runs in the calling thread. If your HTTP thread is busy processing a task, it can't accept new requests. Natural back pressure without data loss. Use this for Kafka consumers and file readers where you can slow down but can't lose work.

// File reader slows itself down automatically
while ((line = reader.readLine()) != null) {
    pool.execute(() -> process(line)); // CallerRunsPolicy throttles the loop
}

**DiscardPolicy** — silently drops the task. Never use this in fintech. Acceptable only where losing 1% of events is fine — analytics, click tracking, activity logs.

**DiscardOldestPolicy** — drops the oldest queued task to make room. Useful for live dashboards and exchange rate feeds where stale data is worse than no data.

Custom policy — the right answer for Revolut:

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    rejectionCount.incrementAndGet();
    log.warn("[REJECTED] pool={} active={} queued={}",
        poolName, e.getActiveCount(), e.getQueue().size());
// fallback chain:
    if (!kafkaFallback.send(r))    // → Kafka retry topic
        if (!dbQueue.save(r))      // → DB if Kafka unavailable
            throw new RejectedExecutionException(); // → last resort
}

Log every rejection. A single rejected payment is a P1 incident.

Factory methods: what they hide

// What the factories actually create:
Executors.newFixedThreadPool(10)
→ new ThreadPoolExecutor(10, 10, 0L, MILLISECONDS, new LinkedBlockingQueue<>())
//                                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^
//                                                   unbounded queue → OOM
Executors.newCachedThreadPool()
→ new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, SECONDS, new SynchronousQueue<>())
//                          ^^^^^^^^^^^^^^^^^
//                          2 billion threads → OOM

Both are traps. The rule: factory methods for prototypes and tests only. ThreadPoolExecutor directly in production.

The interview phrase that lands well: “I know the factories, but in production I always construct ThreadPoolExecutor directly — unbounded queue is a potential OOM, and the default rejection policy doesn't fit fintech."

Future and the problem with get()

Callable<PaymentResult> task = () -> paymentService.process(payment);
Future<PaymentResult> future = pool.submit(task);
// Always with timeout - never block forever
PaymentResult result = future.get(5, TimeUnit.SECONDS);

Future has three problems: get() blocks the calling thread, you can't compose futures, and cancel(true) only works if the task checks Thread.interrupted().

**CompletableFuture is the right path:**

CompletableFuture.supplyAsync(() -> paymentService.process(payment), pool)
    .orTimeout(5, TimeUnit.SECONDS)
    .exceptionally(ex -> PaymentResult.failed(ex.getMessage()));

⚠️ Always pass your own pool to supplyAsync. ForkJoinPool.commonPool() is a shared JVM resource — blocking it breaks everything running in the same JVM, including framework internals.

The submit() trap

This one shows up in code review constantly.

// pool.submit(callable) wraps it in new FutureTask(callable)
// Your RejectedExecutionHandler receives a FutureTask, not your class
// So this never works:
if (r instanceof PaymentTask task) { ... } // always false
// Instead, log context where you still have access to your type:
public Future<PaymentResult> submitPayment(PaymentTask task) {
    try {
        return resolvePool(task.type()).submit(task);
    } catch (RejectedExecutionException e) {
        log.error("[REJECTED] id={} type={}", task.id(), task.type());
        throw e;
    }
}

Monitoring

executor.getActiveCount()         // threads currently executing a task
executor.getQueue().size()        // tasks waiting in queue
executor.getCompletedTaskCount()  // total completed since pool creation
executor.getPoolSize()            // current thread count (including idle)
executor.getLargestPoolSize()     // historical thread count peak

Alert thresholds that matter: queue.size() > 80% capacity is a pre-alert. rejectedCount > 0 is a P1 incident — investigate immediately.

Graceful shutdown

private void shutdownPool(ThreadPoolExecutor pool, String name) {
    pool.shutdown(); // stop accepting new tasks, let current ones finish
    try {
        if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
            List<Runnable> lost = pool.shutdownNow(); // interrupt running threads
            log.error("Pool {} lost {} tasks", name, lost.size());
            // → persist lost tasks to DB for recovery
        }
    } catch (InterruptedException e) {
        pool.shutdownNow();
        Thread.currentThread().interrupt(); // restore the flag — callers need to know
    }
}

Two things to remember: shutdownNow() returns tasks still in the queue — tasks already running are not returned and need a separate recovery mechanism. And Thread.currentThread().interrupt() is not optional — without it, the code above you in the call stack has no way to know the thread was interrupted.

The bugs that show up in every code review

Integer division on small machines:

// On a 2-CPU machine: 20 * 4 / 100 = 0 threads → tasks hang forever
int corePoolSize = 20 * poolSize / 100;
// Correct:
int corePoolSize = Math.max(1, 20 * poolSize / 100);

Hardcoded variable instead of parameter:

private void shutdownPool(ThreadPoolExecutor pool, String name) {
    pool.shutdown();
    if (!instantPool.awaitTermination(...)) // bug — should be `pool`
        instantPool.shutdownNow();          // bug — should be `pool`
}

Copy-paste in stats collection:

PoolStats batchStats = new PoolStats(
    ((PaymentRejectionHandler) this.standardPool  // bug — should be batchPool
        .getRejectedExecutionHandler()).getRejectionCount()
);

Overflow queue with no consumer:

// Tasks go into the queue and die there — worse than DiscardPolicy
// because it creates the illusion of safety
private final BlockingQueue<Runnable> overflowQueue = new LinkedBlockingQueue<>();
overflowQueue.offer(r); // nobody reads this

Practical assignment

Rule: say every decision out loud. “I chose X instead of Y because…” — that’s what separates senior from mid-level.

Scenario

Revolut needs a NotificationDispatcher — sends push, email, and SMS notifications to users. Notifications are non-critical: small losses are acceptable, but the system must not crash under load.

enum NotificationType { PUSH, EMAIL, SMS }
record Notification(String userId, NotificationType type, String message) {}
record DispatcherStats(int active, int queued, long completed, int rejected) {
}

class NotificationDispatcher {
    void dispatch(Notification n);    // submit to pool
    void shutdown();                  // graceful, wait up to 10 seconds
    DispatcherStats getStats();       // current pool state
}

Before writing a line, answer these

• Which RejectedExecutionHandler fits notifications — and why not CallerRunsPolicy?
• Should keepAliveTime be short or long for this workload?
• How do you wait for async tasks to complete in tests without Thread.sleep()?
• shutdown() or shutdownNow() — which does the graceful thing here?

Tests — write these first

@Test
void dispatch_shouldAcceptNotificationWithoutException() {
    Notification n = new Notification("user-1", NotificationType.PUSH, "Hello");
    assertDoesNotThrow(() -> dispatcher.dispatch(n));
}

@Test
void dispatch_shouldExecuteTask() throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(1);
    // hint: wrap dispatch to count down the latch on execution
    assertTrue(latch.await(2, TimeUnit.SECONDS));
}
@Test
void getStats_completedShouldIncrementAfterExecution() throws InterruptedException {
    // dispatch 5 tasks, wait for completion, assert completed == 5
}
@Test
void shutdown_shouldRejectNewTasksAfterShutdown() {
    dispatcher.shutdown();
    assertThrows(RejectedExecutionException.class,
        () -> dispatcher.dispatch(new Notification("u", NotificationType.SMS, "x")));
}
@Test
void dispatch_shouldHandleHighLoad() {
    // 1000 tasks in a row - no OOM, no unhandled exceptions
    assertDoesNotThrow(() -> {
        for (int i = 0; i < 1000; i++) {
            try {
                dispatcher.dispatch(new Notification("user-" + i, NotificationType.EMAIL, "msg"));
            } catch (RejectedExecutionException e) {
                // acceptable - system under load
            }
        }
    });
}

Questions to ask yourself after finishing

• Your pool has corePoolSize = 4, maxPoolSize = 8, queue capacity 100. At what exact moment does the 5th thread get created?
• If a notification task throws an unchecked exception and you have no UncaughtExceptionHandler, what happens to the thread? What about the pool size?
• You chose DiscardPolicy for notifications. A product manager asks for delivery confirmation on SMS. What changes?

Key takeaways

Never use factory methods in production. newFixedThreadPool has an unbounded queue. newCachedThreadPool has unbounded threads. Both are OOM traps.

The growth algorithm is not intuitive. New thread → queue → new thread again → reject. maximumPoolSize is only relevant after the queue fills up.

Queue size is an SLA decision. A queue of 1000 tasks at 500ms processing time means 500 seconds of latency. Small queue + fast rejection is better than a large queue hiding a broken system.

**submit() wraps your task in FutureTask.** If you inspect task types in a rejection handler, you'll never see your class. Log at submission, not at rejection.

**Thread.currentThread().interrupt() is mandatory** after catching InterruptedException. Skip it and the caller is flying blind.

Next: Day 6 — CompletableFuture