Fix QueueResizableOpenSearchThreadPoolExecutorTests

There was a race condition in testResizeQueueDown() where depending on
random parameters we could submit up to 1002 tasks into an executor with
a queue size of 900. That introduced a race condition where if the tasks
didn't execute fast enough then a rejected execution exception could
happen and fail the test. The fix is to resize down to a queue size of
1500 to ensure there is enough capacity even if all tasks are submitted
before any can be executed.

And finally I refactored the tests to reduce duplication of code and
ensure the executor gets shutdown properly even in case of a test
failure. This will avoid the spurious thread leak failure if a test case
exits because of a failure.

Signed-off-by: Andrew Ross <andrross@amazon.com>

Author: andrross

server/src/test/java/org/opensearch/common/util/concurrent/QueueResizableOpenSearchThreadPoolExecutorTests.java

@@ -10,6 +10,8 @@
 
 import org.opensearch.common.settings.Settings;
 import org.opensearch.test.OpenSearchTestCase;
+import org.opensearch.threadpool.ThreadPool;
+import org.junit.After;
 
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -23,158 +25,84 @@
  * based on the time taken for each event.
  */
 public class QueueResizableOpenSearchThreadPoolExecutorTests extends OpenSearchTestCase {
-    public void testResizeQueueSameSize() throws Exception {
-        ThreadContext context = new ThreadContext(Settings.EMPTY);
-        ResizableBlockingQueue<Runnable> queue = new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), 2000);
+    private QueueResizableOpenSearchThreadPoolExecutor executor;
+    private ResizableBlockingQueue<Runnable> queue;
+    private int measureWindow;
 
+    private void createExecutor(int queueSize, Function<Runnable, WrappedRunnable> runnableWrapper) {
         int threads = randomIntBetween(1, 10);
-        int measureWindow = randomIntBetween(100, 200);
-        logger.info("--> auto-queue with a measurement window of {} tasks", measureWindow);
-        QueueResizableOpenSearchThreadPoolExecutor executor = new QueueResizableOpenSearchThreadPoolExecutor(
+        measureWindow = randomIntBetween(100, 200);
+        ThreadContext context = new ThreadContext(Settings.EMPTY);
+        this.queue = new ResizableBlockingQueue<>(ConcurrentCollections.newBlockingQueue(), queueSize);
+        this.executor = new QueueResizableOpenSearchThreadPoolExecutor(
             "test-threadpool",
             threads,
             threads,
             1000,
             TimeUnit.MILLISECONDS,
             queue,
-            fastWrapper(),
+            runnableWrapper,
             OpenSearchExecutors.daemonThreadFactory("queuetest"),
             new OpenSearchAbortPolicy(),
             context
         );
         executor.prestartAllCoreThreads();
         logger.info("--> executor: {}", executor);
+    }
+
+    @After
+    public void stopExecutor() {
+        ThreadPool.terminate(executor, 10, TimeUnit.SECONDS);
+    }
+
+    public void testResizeQueueSameSize() throws Exception {
+        createExecutor(2000, fastWrapper());
 
         // Execute a task multiple times that takes 1ms
         assertThat(executor.resize(1000), equalTo(1000));
         executeTask(executor, (measureWindow * 5) + 2);
 
-        assertBusy(() -> { assertThat(queue.capacity(), lessThanOrEqualTo(1000)); });
-        executor.shutdown();
-        executor.awaitTermination(10, TimeUnit.SECONDS);
+        assertBusy(() -> assertThat(queue.capacity(), lessThanOrEqualTo(1000)));
     }
 
     public void testResizeQueueUp() throws Exception {
-        ThreadContext context = new ThreadContext(Settings.EMPTY);
-        ResizableBlockingQueue<Runnable> queue = new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), 2000);
-
-        int threads = randomIntBetween(1, 10);
-        int measureWindow = randomIntBetween(100, 200);
-        logger.info("--> auto-queue with a measurement window of {} tasks", measureWindow);
-        QueueResizableOpenSearchThreadPoolExecutor executor = new QueueResizableOpenSearchThreadPoolExecutor(
-            "test-threadpool",
-            threads,
-            threads,
-            1000,
-            TimeUnit.MILLISECONDS,
-            queue,
-            fastWrapper(),
-            OpenSearchExecutors.daemonThreadFactory("queuetest"),
-            new OpenSearchAbortPolicy(),
-            context
-        );
-        executor.prestartAllCoreThreads();
-        logger.info("--> executor: {}", executor);
-
+        createExecutor(2000, fastWrapper());
         // Execute a task multiple times that takes 1ms
         assertThat(executor.resize(3000), equalTo(3000));
         executeTask(executor, (measureWindow * 5) + 2);
 
-        assertBusy(() -> { assertThat(queue.capacity(), greaterThanOrEqualTo(2000)); });
-        executor.shutdown();
-        executor.awaitTermination(10, TimeUnit.SECONDS);
+        assertBusy(() -> assertThat(queue.capacity(), greaterThanOrEqualTo(2000)));
     }
 
     public void testResizeQueueDown() throws Exception {
-        ThreadContext context = new ThreadContext(Settings.EMPTY);
-        ResizableBlockingQueue<Runnable> queue = new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), 2000);
-
-        int threads = randomIntBetween(1, 10);
-        int measureWindow = randomIntBetween(100, 200);
-        logger.info("--> auto-queue with a measurement window of {} tasks", measureWindow);
-        QueueResizableOpenSearchThreadPoolExecutor executor = new QueueResizableOpenSearchThreadPoolExecutor(
-            "test-threadpool",
-            threads,
-            threads,
-            1000,
-            TimeUnit.MILLISECONDS,
-            queue,
-            fastWrapper(),
-            OpenSearchExecutors.daemonThreadFactory("queuetest"),
-            new OpenSearchAbortPolicy(),
-            context
-        );
-        executor.prestartAllCoreThreads();
-        logger.info("--> executor: {}", executor);
-
+        createExecutor(2000, fastWrapper());
         // Execute a task multiple times that takes 1ms
-        assertThat(executor.resize(900), equalTo(900));
+        assertThat(executor.resize(1500), equalTo(1500));
         executeTask(executor, (measureWindow * 5) + 2);
 
-        assertBusy(() -> { assertThat(queue.capacity(), lessThanOrEqualTo(900)); });
-        executor.shutdown();
-        executor.awaitTermination(10, TimeUnit.SECONDS);
+        assertBusy(() -> assertThat(queue.capacity(), lessThanOrEqualTo(1500)));
     }
 
     public void testExecutionEWMACalculation() throws Exception {
-        ThreadContext context = new ThreadContext(Settings.EMPTY);
-        ResizableBlockingQueue<Runnable> queue = new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), 100);
-
-        QueueResizableOpenSearchThreadPoolExecutor executor = new QueueResizableOpenSearchThreadPoolExecutor(
-            "test-threadpool",
-            1,
-            1,
-            1000,
-            TimeUnit.MILLISECONDS,
-            queue,
-            fastWrapper(),
-            OpenSearchExecutors.daemonThreadFactory("queuetest"),
-            new OpenSearchAbortPolicy(),
-            context
-        );
-        executor.prestartAllCoreThreads();
-        logger.info("--> executor: {}", executor);
-
+        createExecutor(100, fastWrapper());
         assertThat((long) executor.getTaskExecutionEWMA(), equalTo(0L));
         executeTask(executor, 1);
-        assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(30L)); });
+        assertBusy(() -> assertThat((long) executor.getTaskExecutionEWMA(), equalTo(30L)));
         executeTask(executor, 1);
-        assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(51L)); });
+        assertBusy(() -> assertThat((long) executor.getTaskExecutionEWMA(), equalTo(51L)));
         executeTask(executor, 1);
-        assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(65L)); });
+        assertBusy(() -> assertThat((long) executor.getTaskExecutionEWMA(), equalTo(65L)));
         executeTask(executor, 1);
-        assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(75L)); });
+        assertBusy(() -> assertThat((long) executor.getTaskExecutionEWMA(), equalTo(75L)));
         executeTask(executor, 1);
-        assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(83L)); });
-
-        executor.shutdown();
-        executor.awaitTermination(10, TimeUnit.SECONDS);
+        assertBusy(() -> assertThat((long) executor.getTaskExecutionEWMA(), equalTo(83L)));
     }
 
     /** Use a runnable wrapper that simulates a task with unknown failures. */
-    public void testExceptionThrowingTask() throws Exception {
-        ThreadContext context = new ThreadContext(Settings.EMPTY);
-        ResizableBlockingQueue<Runnable> queue = new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), 100);
-
-        QueueResizableOpenSearchThreadPoolExecutor executor = new QueueResizableOpenSearchThreadPoolExecutor(
-            "test-threadpool",
-            1,
-            1,
-            1000,
-            TimeUnit.MILLISECONDS,
-            queue,
-            exceptionalWrapper(),
-            OpenSearchExecutors.daemonThreadFactory("queuetest"),
-            new OpenSearchAbortPolicy(),
-            context
-        );
-        executor.prestartAllCoreThreads();
-        logger.info("--> executor: {}", executor);
-
+    public void testExceptionThrowingTask() {
+        createExecutor(100, exceptionalWrapper());
         assertThat((long) executor.getTaskExecutionEWMA(), equalTo(0L));
         executeTask(executor, 1);
-        executor.shutdown();
-        executor.awaitTermination(10, TimeUnit.SECONDS);
     }
 
     private Function<Runnable, WrappedRunnable> fastWrapper() {
@@ -198,7 +126,7 @@ private void executeTask(QueueResizableOpenSearchThreadPoolExecutor executor, in
         }
     }
 
-    public class SettableTimedRunnable extends TimedRunnable {
+    private static class SettableTimedRunnable extends TimedRunnable {
         private final long timeTaken;
         private final boolean testFailedOrRejected;