[Feature Request] Add support for a ForkJoinPool type

CHANGELOG.md

@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ## [Unreleased 3.x]
 ### Added
+
+- Add support for a ForkJoinPool type ([#19008](https://github.com/opensearch-project/OpenSearch/pull/19008))
 - Add seperate shard limit validation for local and remote indices ([#19532](https://github.com/opensearch-project/OpenSearch/pull/19532))
 - Use Lucene `pack` method for `half_float` and `usigned_long` when using `ApproximatePointRangeQuery`.
 - Add a mapper for context aware segments grouping criteria ([#19233](https://github.com/opensearch-project/OpenSearch/pull/19233))

qa/evil-tests/src/test/java/org/opensearch/threadpool/EvilThreadPoolTests.java

@@ -47,6 +47,8 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ForkJoinPool;
+
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
@@ -70,6 +72,11 @@ public void tearDownThreadPool() {
 
     public void testExecutionErrorOnDefaultThreadPoolTypes() throws InterruptedException {
         for (String executor : ThreadPool.THREAD_POOL_TYPES.keySet()) {
+            // ForkJoinPool is skipped here because it does not support all ThreadPoolExecutor features or APIs,
+            // and is tested separately in testExecutionErrorOnForkJoinPool.
+            if (ThreadPool.THREAD_POOL_TYPES.get(executor) == ThreadPool.ThreadPoolType.FORK_JOIN) {
+                continue; // skip FORK_JOIN for these tests
+            }
             checkExecutionError(getExecuteRunner(threadPool.executor(executor)));
             checkExecutionError(getSubmitRunner(threadPool.executor(executor)));
             checkExecutionError(getScheduleRunner(executor));
@@ -176,6 +183,11 @@ protected void doRun() {
 
     public void testExecutionExceptionOnDefaultThreadPoolTypes() throws InterruptedException {
         for (String executor : ThreadPool.THREAD_POOL_TYPES.keySet()) {
+            // ForkJoinPool is skipped here because it does not support all ThreadPoolExecutor features or APIs,
+            // and is tested separately in testExecutionErrorOnForkJoinPool.
+            if (ThreadPool.THREAD_POOL_TYPES.get(executor) == ThreadPool.ThreadPoolType.FORK_JOIN) {
+                continue; // skip FORK_JOIN for these tests
+            }
             checkExecutionException(getExecuteRunner(threadPool.executor(executor)), true);
 
             // here, it's ok for the exception not to bubble up. Accessing the future will yield the exception
@@ -391,4 +403,43 @@ private void runExecutionTest(
         }
     }
 
+    public void testExecutionExceptionOnForkJoinPool() throws InterruptedException {
+        ForkJoinPool fjp = new ForkJoinPool();
+        try {
+            checkExecutionException(getExecuteRunner(fjp), true);
+            checkExecutionException(getSubmitRunner(fjp), false);
+        } finally {
+            fjp.shutdownNow();
+            fjp.awaitTermination(10, TimeUnit.SECONDS);
+        }
+    }
+
+    public void testExecutionErrorOnForkJoinPool() throws Exception {
+        ForkJoinPool fjp = new ForkJoinPool(8);
+        final CountDownLatch latch = new CountDownLatch(1);
+        final AtomicReference<Throwable> thrown = new AtomicReference<>();
+        try {
+            fjp.execute(() -> {
+                try {
+                    throw new Error("future error");
+                } catch (Throwable t) {
+                    thrown.set(t);
+                } finally {
+                    latch.countDown();
+                }
+            });
+
+            // Wait up to 5 seconds for the task to complete
+            assertTrue("Timeout waiting for ForkJoinPool task", latch.await(5, TimeUnit.SECONDS));
+
+            Throwable error = thrown.get();
+            assertNotNull("No error captured from ForkJoinPool task", error);
+            assertTrue(error instanceof Error);
+            assertEquals("future error", error.getMessage());
+        } finally {
+            fjp.shutdownNow();
+            fjp.awaitTermination(10, TimeUnit.SECONDS);
+        }
+    }
+
 }

rest-api-spec/src/main/resources/rest-api-spec/test/cat.thread_pool/10_basic.yml

@@ -60,72 +60,72 @@
 
   - match:
       $body: |
-               /  #node_name     name     active     queue     rejected
-               ^  (\S+       \s+ \S+  \s+ \d+    \s+ \d+   \s+ \d+      \n)+  $/
+        /  #node_name     name     active     queue     rejected
+        ^  (\S+       \s+ \S+  \s+ \d+    \s+ \d+   \s+ \d+      \n)+  $/
 
   - do:
       cat.thread_pool:
-          v: true
+        v: true
 
   - match:
-        $body: |
-                 /^   node_name \s+ name \s+ active \s+ queue \s+ rejected \n
-                     (\S+       \s+ \S+  \s+ \d+    \s+ \d+   \s+ \d+      \n)+  $/
+      $body: |
+        /^   node_name \s+ name \s+ active \s+ queue \s+ rejected \n
+            (\S+       \s+ \S+  \s+ \d+    \s+ \d+   \s+ \d+      \n)+  $/
 
   - do:
       cat.thread_pool:
-          h: pid,id,h,i,po
+        h: pid,id,h,i,po
 
   - match:
       $body: |
-               /  #pid       id          host       ip                          port
-                  (\d+  \s+  \S+    \s+  \S+   \s+  (\d{1,3}\.){3}\d{1,3}  \s+  (\d+|-)  \n)+  $/
+        /  #pid       id          host       ip                          port
+           (\d+  \s+  \S+    \s+  \S+   \s+  (\d{1,3}\.){3}\d{1,3}  \s+  (\d+|-)  \n)+  $/
 
   - do:
       cat.thread_pool:
-          thread_pool_patterns: write,management,flush,generic,force_merge
-          h: id,name,active
-          v: true
+        thread_pool_patterns: write,management,flush,generic,force_merge
+        h: id,name,active
+        v: true
 
   - match:
       $body: |
-               /^  id \s+ name        \s+ active \n
-                  (\S+\s+ flush       \s+ \d+    \n
-                   \S+\s+ force_merge \s+ \d+    \n
-                   \S+\s+ generic     \s+ \d+    \n
-                   \S+\s+ management  \s+ \d+    \n
-                   \S+\s+ write       \s+ \d+    \n)+  $/
+        /^  id \s+ name        \s+ active \n
+           (\S+\s+ flush       \s+ \d+    \n
+            \S+\s+ force_merge \s+ \d+    \n
+            \S+\s+ generic     \s+ \d+    \n
+            \S+\s+ management  \s+ \d+    \n
+            \S+\s+ write       \s+ \d+    \n)+  $/
 
   - do:
       cat.thread_pool:
-          thread_pool_patterns: write
-          h: id,name,type,active,size,queue,queue_size,rejected,largest,completed,min,max,keep_alive
-          v: true
+        thread_pool_patterns: write
+        h: id,name,type,active,size,queue,queue_size,rejected,largest,completed,min,max,keep_alive
+        v: true
 
   - match:
       $body: |
-               /^  id  \s+ name  \s+ type  \s+ active \s+ size \s+ queue \s+ queue_size \s+ rejected \s+ largest \s+ completed \s+ max \s+ keep_alive \n
-                  (\S+ \s+ write \s+ fixed \s+ \d+    \s+ \d+  \s+ \d+   \s+ (-1|\d+)   \s+ \d+      \s+ \d+     \s+ \d+       \s+ \d* \s+ \S*        \n)+  $/
+        /^  id  \s+ name  \s+ type  \s+ active \s+ size \s+ queue \s+ queue_size \s+ rejected \s+ largest \s+ completed \s+ max \s+ keep_alive \n
+           (\S+ \s+ write \s+ fixed \s+ \d+    \s+ \d+  \s+ \d+   \s+ (-1|\d+)   \s+ \d+      \s+ \d+     \s+ \d+       \s+ \d* \s+ \S*        \n)+  $/
 
   - do:
       cat.thread_pool:
-          thread_pool_patterns: fetch*
-          h: id,name,type,active,pool_size,queue,queue_size,rejected,largest,completed,core,max,size,keep_alive
-          v: true
+        thread_pool_patterns: fetch*
+        h: id,name,type,active,pool_size,queue,queue_size,rejected,largest,completed,core,max,size,keep_alive
+        v: true
 
   - match:
       $body: |
-               /^  id  \s+ name                \s+ type    \s+ active \s+ pool_size \s+ queue \s+ queue_size \s+ rejected \s+ largest \s+ completed \s+ core \s+ max \s+ size \s+ keep_alive \n
-                  (\S+ \s+ fetch_shard_started \s+ scaling \s+ \d+    \s+ \d+       \s+ \d+   \s+ (-1|\d+)   \s+ \d+      \s+ \d+     \s+ \d+       \s+ \d*  \s+ \d* \s+ \d*  \s+ \S*        \n
-                   \S+ \s+ fetch_shard_store   \s+ scaling \s+ \d+    \s+ \d+       \s+ \d+   \s+ (-1|\d+)   \s+ \d+      \s+ \d+     \s+ \d+       \s+ \d*  \s+ \d* \s+ \d*  \s+ \S*        \n)+  $/
+        /^  id  \s+ name                \s+ type    \s+ active \s+ pool_size \s+ queue \s+ queue_size \s+ rejected \s+ largest \s+ completed \s+ core \s+ max \s+ size \s+ keep_alive \n
+           (\S+ \s+ fetch_shard_started \s+ scaling \s+ \d+    \s+ \d+       \s+ \d+   \s+ (-1|\d+)   \s+ \d+      \s+ \d+     \s+ \d+       \s+ \d*  \s+ \d* \s+ \d*  \s+ \S*        \n
+            \S+ \s+ fetch_shard_store   \s+ scaling \s+ \d+    \s+ \d+       \s+ \d+   \s+ (-1|\d+)   \s+ \d+      \s+ \d+     \s+ \d+       \s+ \d*  \s+ \d* \s+ \d*  \s+ \S*        \n)+  $/
 
   - do:
       cat.thread_pool:
-          thread_pool_patterns: write,search
-          size: ""
+        thread_pool_patterns: write,search
+        size: ""
 
   - match:
       $body: |
-               /  #node_name     name       active     queue     rejected
-               ^  (\S+       \s+ search  \s+ \d+    \s+ \d+   \s+ \d+      \n
-                   \S+       \s+ write   \s+ \d+    \s+ \d+   \s+ \d+      \n)+  $/
+        /  #node_name     name       active     queue     rejected
+        ^  (\S+       \s+ search  \s+ \d+    \s+ \d+   \s+ \d+      \n
+            \S+       \s+ write   \s+ \d+    \s+ \d+   \s+ \d+      \n)+  $/

server/src/internalClusterTest/java/org/opensearch/threadpool/ForkJoinPoolIT.java

@@ -0,0 +1,58 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.threadpool;
+
+import org.opensearch.common.settings.Settings;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.test.OpenSearchSingleNodeTestCase;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ForkJoinPool;
+
+/**
+ * Single-node IT that defines an inline plugin to register a ForkJoin executor ("jvector")
+ * and verifies it is available on the node.
+ */
+public class ForkJoinPoolIT extends OpenSearchSingleNodeTestCase {
+
+    /**
+     * Inline test plugin that registers a ForkJoin-based executor named "jvector"
+     * with a fixed parallelism of 9 for deterministic assertions.
+     */
+    public static class TestPlugin extends Plugin {
+        @Override
+        public List<ExecutorBuilder<?>> getExecutorBuilders(final Settings settings) {
+            return List.of(new ForkJoinPoolExecutorBuilder("jvector", 9));
+        }
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> getPlugins() {
+        // Load the inline plugin into the single-node cluster for this test
+        return List.of(TestPlugin.class);
+    }
+
+    public void testForkJoinPoolExists() {
+        // Obtain the node's ThreadPool and verify the "jvector" executor
+        ThreadPool threadPool = getInstanceFromNode(ThreadPool.class);
+        ExecutorService executor = threadPool.executor("jvector");
+        assertNotNull("jvector executor should be registered by the test plugin", executor);
+        assertTrue("jvector should be a ForkJoinPool", executor instanceof ForkJoinPool);
+        assertEquals("parallelism should be 9", 9, ((ForkJoinPool) executor).getParallelism());
+
+        // Also validate ThreadPool.Info reports FORK_JOIN with expected parallelism (max)
+        ThreadPool.Info info = threadPool.info("jvector");
+        assertNotNull("ThreadPool.Info for jvector should exist", info);
+        assertEquals("jvector", info.getName());
+        assertEquals("type must be FORK_JOIN", ThreadPool.ThreadPoolType.FORK_JOIN, info.getThreadPoolType());
+        assertEquals("info.max should equal parallelism", 9, info.getMax());
+    }
+}