[Feature Request] Add support for a ForkJoinPool type

CHANGELOG.md

@@ -5,6 +5,7 @@ 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))
 - Expand fetch phase profiling to support inner hits and top hits aggregation phases  ([##18936](https://github.com/opensearch-project/OpenSearch/pull/18936))
 - Add temporal routing processors for time-based document routing ([#18920](https://github.com/opensearch-project/OpenSearch/issues/18920))
 - Implement Query Rewriting Infrastructure ([#19060](https://github.com/opensearch-project/OpenSearch/pull/19060))

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);
+        }
+    }
+
 }

qa/mixed-cluster/build.gradle

@@ -40,6 +40,7 @@ apply plugin: 'opensearch.rest-resources'
 
 dependencies {
   testImplementation project(":client:rest-high-level")
+  testImplementation project(":libs:opensearch-common")
 }
 
 restResources {

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

@@ -1,7 +1,7 @@
 "Test cat thread_pool total_wait_time output":
   - skip:
-      version: " - 2.10.99"
-      reason: thread_pool total_wait_time stats were introduced in V_2.11.0
+      version: " - 2.10.99,3.2.0 -"
+      reason: thread_pool total_wait_time stats were introduced in V_2.11.0; ForkJoinPool thread pool support was introduced in 3.2.0 and is tested separately.
 
   - do:
       cat.thread_pool: {}
@@ -27,8 +27,8 @@
 ---
 "Test cat thread_pool total_wait_time output with concurrent search thread_pool":
   - skip:
-      version: " - 2.11.99"
-      reason: index_search thread_pool was introduced in V_2.12.0
+      version: " - 2.11.99,3.2.0 -"
+      reason: index_search thread_pool was introduced in V_2.12.0; ForkJoinPool thread pool support was introduced in 3.2.0 and is tested separately.
 
   - do:
       cat.thread_pool: {}
@@ -54,6 +54,9 @@
 
 ---
 "Test cat thread_pool output":
+  - skip:
+      version: "3.2.0 -,"
+      reason: ForkJoinPool thread pool support was introduced in 3.2.0 and is tested separately.
 
   - do:
       cat.thread_pool: {}

rest-api-spec/src/main/resources/rest-api-spec/test/cat.thread_pool/20_fork_join_pool.yml

@@ -0,0 +1,99 @@
+---
+"Test cat thread_pool fork_join output":
+  - skip:
+      version: " - 3.1.99"
+      reason: fork_join thread pool support was introduced in 3.2.0
+  - do:
+      cat.thread_pool:
+        thread_pool_patterns: fork_join
+        h: node_name,name,type,active,queue,queue_size,rejected,largest,completed,max,keep_alive
+        v: true
+  # At least one row must contain 'fork_join' in name, optionally in type, and 0 for active, -1 for queue_size
+  - match:
+      $body: /fork_join.*0.*-1/
+
+---
+"Test cat thread_pool fork_join with unrelated pool":
+  - skip:
+      version: " - 3.1.99"
+      reason: fork_join thread pool support was introduced in 3.2.0
+  - do:
+      cat.thread_pool:
+        thread_pool_patterns: fork_join,search
+        h: name,type,active,queue,queue_size,rejected
+        v: true
+  # At least one fork_join row must exist
+  - match:
+      $body: /fork_join.*0.*-1/
+  # At least one search row, queue_size may be -1 or 1000, type can be blank or 'resizable'
+  - match:
+      $body: /search.*0.*(-1|1000)/
+
+---
+"Test cat thread_pool fork_join with only fork_join present":
+  - skip:
+      version: " - 3.1.99"
+      reason: fork_join thread pool support was introduced in 3.2.0
+  - do:
+      cat.thread_pool:
+        thread_pool_patterns: fork_join
+        h: name,type,active,queue,queue_size,rejected
+        v: true
+  - match:
+      $body: /fork_join.*0.*-1/
+
+---
+"Test cat thread_pool fork_join with custom header order":
+  - skip:
+      version: " - 3.1.99"
+      reason: fork_join thread pool support was introduced in 3.2.0
+  - do:
+      cat.thread_pool:
+        thread_pool_patterns: fork_join
+        h: queue,active,name,type
+        v: true
+  # Accept row with 0 queue, 0 active, fork_join in name, type may appear or not
+  - match:
+      $body: /0.*0.*fork_join/
+
+---
+"Test cat thread_pool fork_join missing optional columns":
+  - skip:
+      version: " - 3.1.99"
+      reason: fork_join thread pool support was introduced in 3.2.0
+  - do:
+      cat.thread_pool:
+        thread_pool_patterns: fork_join
+        h: name,type,keep_alive,queue_size,max
+        v: true
+  # At least one fork_join row with -1 in keep_alive
+  - match:
+      $body: /fork_join.*-1/
+
+---
+"Test cat thread_pool fork_join with unknown header":
+  - skip:
+      version: " - 3.1.99"
+      reason: fork_join thread pool support was introduced in 3.2.0
+  - do:
+      cat.thread_pool:
+        thread_pool_patterns: fork_join
+        h: name,type,foo,active
+        v: true
+  # At least one fork_join row with 0 in active
+  - match:
+      $body: /fork_join.*0/
+
+---
+"Test cat thread_pool fork_join with v:false":
+  - skip:
+      version: " - 3.1.99"
+      reason: fork_join thread pool support was introduced in 3.2.0
+  - do:
+      cat.thread_pool:
+        thread_pool_patterns: fork_join
+        h: name,type,active
+        v: false
+  # At least one row with fork_join and 0 in active
+  - match:
+      $body: /fork_join.*0/

server/src/main/java/org/opensearch/rest/action/cat/RestThreadPoolAction.java

@@ -175,6 +175,8 @@ protected Table getTableWithHeader(final RestRequest request) {
         return table;
     }
 
+    // ... [imports and class definition unchanged] ...
+
     private Table buildTable(RestRequest req, ClusterStateResponse state, NodesInfoResponse nodesInfo, NodesStatsResponse nodesStats) {
         final String[] threadPools = req.paramAsStringArray("thread_pool_patterns", new String[] { "*" });
         final DiscoveryNodes nodes = state.getState().nodes();
@@ -196,6 +198,7 @@ private Table buildTable(RestRequest req, ClusterStateResponse state, NodesInfoR
             }
         }
 
+        Set<String> forkJoinEmitted = new HashSet<>();
         for (final DiscoveryNode node : nodes) {
             final NodeInfo info = nodesInfo.getNodesMap().get(node.getId());
             final NodeStats stats = nodesStats.getNodesMap().get(node.getId());
@@ -225,6 +228,44 @@ private Table buildTable(RestRequest req, ClusterStateResponse state, NodesInfoR
 
                 if (!included.contains(entry.getKey())) continue;
 
+                final ThreadPoolStats.Stats poolStats = entry.getValue();
+                final ThreadPool.Info poolInfo = poolThreadInfo.get(entry.getKey());
+
+                boolean isForkJoin = poolInfo != null && poolInfo.getThreadPoolType() == ThreadPool.ThreadPoolType.FORK_JOIN;
+                if (isForkJoin) {
+                    // only emit a single row for fork_join for the whole cluster
+                    if (forkJoinEmitted.contains(entry.getKey())) continue;
+                    forkJoinEmitted.add(entry.getKey());
+
+                    table.startRow();
+
+                    table.addCell(node.getName());
+                    table.addCell(node.getId());
+                    table.addCell(node.getEphemeralId());
+                    table.addCell(info == null ? null : info.getInfo(ProcessInfo.class).getId());
+                    table.addCell(node.getHostName());
+                    table.addCell(node.getHostAddress());
+                    table.addCell(node.getAddress().address().getPort());
+                    // pool columns
+                    table.addCell(entry.getKey());
+                    table.addCell(poolInfo.getThreadPoolType().getType());
+                    // ForkJoinPool: most stats undefined or 0/-1
+                    table.addCell(0);      // active
+                    table.addCell(0);      // pool_size
+                    table.addCell(0);      // queue
+                    table.addCell(-1);     // queue_size
+                    table.addCell(0);      // rejected
+                    table.addCell(0);      // largest
+                    table.addCell(0);      // completed
+                    table.addCell(-1);     // total_wait_time
+                    table.addCell(null);   // core
+                    table.addCell(null);   // max
+                    table.addCell(null);   // size
+                    table.addCell(null);   // keep_alive
+                    table.endRow();
+                    continue;
+                }
+
                 table.startRow();
 
                 table.addCell(node.getName());
@@ -234,8 +275,6 @@ private Table buildTable(RestRequest req, ClusterStateResponse state, NodesInfoR
                 table.addCell(node.getHostName());
                 table.addCell(node.getHostAddress());
                 table.addCell(node.getAddress().address().getPort());
-                final ThreadPoolStats.Stats poolStats = entry.getValue();
-                final ThreadPool.Info poolInfo = poolThreadInfo.get(entry.getKey());
 
                 Long maxQueueSize = null;
                 String keepAlive = null;

server/src/main/java/org/opensearch/threadpool/ForkJoinPoolExecutorBuilder.java

@@ -0,0 +1,81 @@
+/*
+ * 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.Setting;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.concurrent.OpenSearchExecutors;
+import org.opensearch.common.util.concurrent.ThreadContext;
+import org.opensearch.node.Node;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinPool.ForkJoinWorkerThreadFactory;
+import java.util.concurrent.ForkJoinWorkerThread;
+
+/**
+ * A builder for fork join executors.
+ *
+ * @opensearch.internal
+ */
+public final class ForkJoinPoolExecutorBuilder extends ExecutorBuilder<ForkJoinPoolExecutorBuilder.ForkJoinPoolExecutorSettings> {
+
+    private final Setting<Integer> parallelismSetting;
+
+    public ForkJoinPoolExecutorBuilder(final String name, final int parallelism) {
+        this(name, parallelism, "thread_pool." + name);
+    }
+
+    public ForkJoinPoolExecutorBuilder(final String name, final int parallelism, final String prefix) {
+        super(name);
+        this.parallelismSetting = Setting.intSetting(settingsKey(prefix, "parallelism"), parallelism, Setting.Property.NodeScope);
+    }
+
+    @Override
+    public List<Setting<?>> getRegisteredSettings() {
+        return Arrays.asList(parallelismSetting);
+    }
+
+    @Override
+    ForkJoinPoolExecutorSettings getSettings(Settings settings) {
+        final String nodeName = Node.NODE_NAME_SETTING.get(settings);
+        final int parallelism = parallelismSetting.get(settings);
+        return new ForkJoinPoolExecutorSettings(nodeName, parallelism);
+    }
+
+    @Override
+    ThreadPool.ExecutorHolder build(final ForkJoinPoolExecutorSettings settings, final ThreadContext threadContext) {
+        int parallelism = settings.parallelism;
+        ForkJoinWorkerThreadFactory factory = pool -> {
+            ForkJoinWorkerThread worker = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
+            worker.setName(OpenSearchExecutors.threadName(settings.nodeName, name()));
+            return worker;
+        };
+        final ForkJoinPool executor = new ForkJoinPool(parallelism, factory, null, false);
+
+        final ThreadPool.Info info = new ThreadPool.Info(name(), ThreadPool.ThreadPoolType.FORK_JOIN, parallelism, parallelism, null, null);
+        return new ThreadPool.ExecutorHolder(executor, info);
+    }
+
+    @Override
+    String formatInfo(ThreadPool.Info info) {
+        return String.format(Locale.ROOT, "name [%s], parallelism [%d]", info.getName(), info.getMax());
+    }
+
+    static class ForkJoinPoolExecutorSettings extends ExecutorBuilder.ExecutorSettings {
+        private final int parallelism;
+
+        ForkJoinPoolExecutorSettings(final String nodeName, final int parallelism) {
+            super(nodeName);
+            this.parallelism = parallelism;
+        }
+    }
+}