Java自定义ThreadPool—暂停任务提交并取消当前排队任务

我有以下场景:一个自定义线程池需要能够暂停.暂停时，删除当前排队的任务.在暂停期间，它可以接受任务，并在恢复时执行"新"排队任务.我目前的实现就是在做上述要求.线程池实现:

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class QueriesThreadPoolExecutor extends ThreadPoolExecutor {
    private final List<Future<?>> futuresList = Collections.synchronizedList(new ArrayList<>());
    private boolean isPaused = false;
    private Lock pauseLock = new ReentrantLock();
    private Condition unpaused = pauseLock.newCondition();

    public QueriesThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    @Override
    public void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
//        System.out.println("beforeExecute " + t.getName());
        pauseLock.lock();
        try {
            while (isPaused) {
                unpaused.await();
            }
        } catch (InterruptedException e) {
            t.interrupt();
        } finally {
            pauseLock.unlock();
        }
    }

    @Override
    public void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
//        System.out.println("afterExecute " + Thread.currentThread().getName());
    }

    @Override
    public void terminated() {
        super.terminated();
        System.out.println("Thread pool terminated");
    }

    public List<Runnable> shutdownNow() {
        List<Runnable> tasks = super.shutdownNow();

        System.out.println("Shutting down thread pool, active tasks remaining: " + tasks.size());
        return tasks;
    }

    public <T> Future<T> submit(Callable<T> task) {
        System.out.println("submit task");
        Future<T> res = super.submit(task);
        futuresList.add(res);

        return res;
    }

    public void pause() {
        cancelRunningTasks();

        System.out.println("pause");
        pauseLock.lock();
        try {
            isPaused = true;
        } finally {
            pauseLock.unlock();
        }
    }

    public void resume() {
        System.out.println("resume");
        pauseLock.lock();
        try {
            isPaused = false;
            unpaused.signal();
        } finally {
            pauseLock.unlock();
        }
    }

    public void cancelRunningTasks() {
        synchronized (futuresList) {
            for (Future<?> future : futuresList) {
                if (!future.isDone()) {
                    future.cancel(true);
                    System.out.println("cancelRunningTasks");
                }
            }
        }
    }
}

单元测试

import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

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

class QueriesThreadPoolExecutorTests {
    private static QueriesThreadPoolExecutor queriesThreadPoolExecutor;
    private AtomicInteger longCounter = new AtomicInteger(0);
    private AtomicInteger shortCounter = new AtomicInteger(0);

    @BeforeAll
    static void setup() {
        queriesThreadPoolExecutor = new QueriesThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
    }

    @AfterAll
    static void tearDown() {
        queriesThreadPoolExecutor.shutdownNow();
    }

    @Test
    void testCancelResumeTasks() {
        Callable<String> longRunningTask = () -> {
            Thread.sleep(20_000);
            return "long running task" + longCounter.incrementAndGet();
        };

        Callable<String> shortRunningTask = () -> {
            Thread.sleep(1_000);
            return "short running task" + shortCounter.incrementAndGet();
        };
        List<Future<String>> futureList = new ArrayList<>();

        submitTasks(futureList, shortRunningTask, longRunningTask);

        ScheduledExecutorService cancelScheduledExecutor = Executors.newScheduledThreadPool(2);
        cancelScheduledExecutor.schedule(() -> {
            queriesThreadPoolExecutor.pause();
        }, 2, TimeUnit.SECONDS);

        printResults(futureList);

        submitTasks(futureList, shortRunningTask, longRunningTask);

        queriesThreadPoolExecutor.resume();

        printResults(futureList);
    }

    private void submitTasks(List<Future<String>> futureList, Callable<String> shortRunningTask, Callable<String> longRunningTask) {
        for (int i = 0; i < 5; i++) {
            futureList.add(queriesThreadPoolExecutor.submit(shortRunningTask));
        }
        for (int i = 0; i < 5; i++) {
            futureList.add(queriesThreadPoolExecutor.submit(longRunningTask));
        }
    }

    private void printResults(List<Future<String>> futureList) {
        System.out.println("***********Printing results*****************");
        for (Future<String> future : futureList) {
            try {
                String status = future.isCancelled() ? "cancelled" : "done";
                String result = future.isCancelled() ? "cancelled" : future.get();
                System.out.println("task status - " + status + " - result " + result);
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            } catch (CancellationException e) {
                System.out.println("Exception - Task was cancelled");
            }
        }
    }
}

由于某种原因，当线程池暂停时，一个或多个任务从单元测试中引发异常:

task status - done - result short running task1
cancelRunningTasks
cancelRunningTasks
cancelRunningTasks
cancelRunningTasks
cancelRunningTasks
Exception - Task was cancelled  <--------------------WHY?
task status - cancelled - result cancelled
task status - cancelled - result cancelled
task status - cancelled - result cancelled
task status - cancelled - result cancelled
submit task
pause
submit task

也许我错过了一些琐碎的东西.