Understanding Java ThreadPoolExecutor Rejection Policies and How to Prevent Task Loss

Core Mechanism of ThreadPoolExecutor Rejection

ThreadPoolExecutor invokes a RejectedExecutionHandler when either the pool is saturated (active thread count ≥ maximumPoolSize or the bounded workQueue is full) or the pool has been shut down via shutdown() and a new task is submitted.

Task submission calls execute() to check pool state.

If the pool cannot accept the task, RejectedExecutionHandler.rejectedExecution() is executed.

Four Built‑in Rejection Policies

AbortPolicy : throws RejectedExecutionException. Suitable for critical tasks (e.g., payment). Caller must handle the exception.

CallerRunsPolicy : the submitting thread runs the task. Suitable for non‑critical tasks that must be executed (e.g., log reporting). May block the caller thread.

DiscardPolicy : silently discards the task. Suitable for tasks that can be lost (e.g., monitoring data). Risk of task loss.

DiscardOldestPolicy : discards the oldest queued task and retries the new one. Suitable for high‑real‑time requirements (e.g., stock quotes). May lose important tasks.

Solutions to Ensure No Task Loss

Custom Rejection Handler with Persistent Storage

Core idea: serialize the rejected Runnable and store it in external storage (database, message queue, or file). A scheduled retry job reads pending records, deserializes them, re‑submits to the executor, and updates their status.

Implementation steps:

Define a persisted task entity:

@Data
public class PersistedTask {
    private String id;
    private String taskData; // serialized task
    private int retryCount;
    private LocalDateTime createTime;
}

Create a custom RejectedExecutionHandler that serializes and saves the task:

public class PersistenceRejectPolicy implements RejectedExecutionHandler {
    private final TaskRepository taskRepository;
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        try {
            String taskData = serializeTask(r);
            taskRepository.save(new PersistedTask(UUID.randomUUID().toString(), taskData, 0));
        } catch (Exception e) {
            log.error("Persist task failed", e);
        }
    }
    private String serializeTask(Runnable r) {
        return new Gson().toJson(r);
    }
}

Schedule a retry job that processes pending tasks:

@Scheduled(fixedRate = 5000)
public void retryRejectedTasks() {
    List<PersistedTask> tasks = taskRepository.findPendingTasks();
    for (PersistedTask task : tasks) {
        try {
            Runnable r = deserializeTask(task.getTaskData());
            executor.execute(r);
            taskRepository.markAsCompleted(task.getId());
        } catch (Exception e) {
            if (task.getRetryCount() >= 3) {
                taskRepository.markAsFailed(task.getId());
            } else {
                taskRepository.incrementRetry(task.getId());
            }
        }
    }
}

Asynchronous Handling with a Message Queue

Decouple submission from execution by sending the serialized task to a Kafka topic.

public class MqRejectPolicy implements RejectedExecutionHandler {
    private final KafkaTemplate<String, String> kafkaTemplate;
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        kafkaTemplate.send("rejected-tasks", serializeTask(r));
    }
}

Thread‑Pool Parameter Optimization

Queue selection: use a bounded ArrayBlockingQueue with capacity = maxThreads * 2, or a PriorityBlockingQueue when task ordering is required.

Thread count: set corePoolSize = CPU cores for CPU‑bound work; set corePoolSize = 2 * CPU cores for I/O‑bound work.

Graceful Shutdown

public void shutdownGracefully(ThreadPoolExecutor executor) {
    executor.shutdown(); // reject new tasks
    try {
        if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
            List<Runnable> pendingTasks = executor.shutdownNow(); // attempt to stop running tasks
            savePendingTasks(pendingTasks); // persist unfinished tasks
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

Practical Scenarios

Scenario 1: High‑Concurrency Order Processing

Policy: AbortPolicy + PersistenceRejectPolicy + circuit‑breaker.

Configuration example:

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    50, 200, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(1000),
    new PersistenceRejectPolicy(taskRepository));

Degradation: when the database connection pool is exhausted, the circuit‑breaker discards non‑critical orders.

Scenario 2: Real‑Time Data Analysis

Policy: DiscardOldestPolicy + MqRejectPolicy (Kafka).

Configuration example:

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10, 50, 30, TimeUnit.SECONDS,
    new PriorityBlockingQueue<>(100),
    new MqRejectPolicy(kafkaTemplate));

Risks and Optimization Recommendations

Persistence performance bottleneck : use batch inserts or a Redis List as a temporary store.

Task duplicate execution : assign a unique ID to each task and check before processing.

Memory leak : periodically clean up back‑log tasks in an emergency queue.

Missing monitoring : expose metrics threadpool.rejected.count and threadpool.queue.size via Micrometer.

Conclusion

Preventing task loss requires selecting an appropriate rejection policy (built‑in or custom persistence), designing a reliable persistence layer (database or Kafka), implementing a retry mechanism, tuning thread‑pool parameters, and adding proper monitoring and fallback strategies.