Initial chekpoint Feature Request - Add support for a ForkJoinPool type

Signed-off-by: Srikanth Padakanti <[email protected]>

Author: Srikanth Padakanti

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

@@ -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; fork_join thread pool not present before 3.2.0
 
   - do:
       cat.thread_pool: {}
@@ -129,3 +129,32 @@
                /  #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)+  $/
+
+# NEW TEST for 3.2.0 and later (expect fork_join)
+---
+"Test cat thread_pool total_wait_time output (with fork_join)":
+  - skip:
+      version: " - 3.1.99"
+      reason: fork_join thread pool introduced in 3.2.0
+
+  - do:
+      cat.thread_pool: {}
+
+  - match:
+      $body: |
+        /  #node_name     name     active     queue     rejected
+        ^  (\S+       \s+ \S+  \s+ \d+    \s+ \d+   \s+ \d+      \n)+  $/
+
+  - do:
+      cat.thread_pool:
+        thread_pool_patterns: search,search_throttled,generic,fork_join
+        h: name,total_wait_time,twt
+        v: true
+
+  - match:
+      $body: |
+        /^  name              \s+ total_wait_time \s+ twt              \n
+           (generic           \s+ -1              \s+ -1               \n
+            fork_join         \s+ -1              \s+ -1               \n
+            search            \s+ \d*\.*\d*\D+    \s+ \d*\.*\d*\D+     \n
+            search_throttled  \s+ \d*\.*\d*\D+    \s+ \d*\.*\d*\D+     \n)+  $/

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

@@ -262,16 +262,45 @@ private Table buildTable(RestRequest req, ClusterStateResponse state, NodesInfoR
                     }
                 }
 
+                // --- BEGIN CHANGE: Normalize ForkJoinPool stats to 0 ---
+                boolean isForkJoin = poolInfo != null && poolInfo.getThreadPoolType() == ThreadPool.ThreadPoolType.FORK_JOIN;
+
+                int active = (poolStats == null ? 0 : poolStats.getActive());
+                int threads = (poolStats == null ? 0 : poolStats.getThreads());
+                int queue = (poolStats == null ? 0 : poolStats.getQueue());
+                long rejected = (poolStats == null ? 0 : poolStats.getRejected());
+                int largest = (poolStats == null ? 0 : poolStats.getLargest());
+                long completed = (poolStats == null ? 0 : poolStats.getCompleted());
+                long waitTime = (poolStats == null ? 0 : poolStats.getWaitTime().nanos());
+
+                // Normalize ForkJoinPool stats to 0 or -1 as per convention
+                if (isForkJoin) {
+                    active = 0;
+                    threads = 0;
+                    queue = 0;
+                    rejected = 0;
+                    largest = 0;
+                    completed = 0;
+                    waitTime = -1; // if wait time is unsupported
+                    // Set maxQueueSize to -1 if not supported for ForkJoinPool
+                    maxQueueSize = -1L;
+                    core = null;
+                    max = null;
+                    size = null;
+                    keepAlive = null;
+                }
+                // --- END CHANGE ---
+
                 table.addCell(entry.getKey());
                 table.addCell(poolInfo == null ? null : poolInfo.getThreadPoolType().getType());
-                table.addCell(poolStats == null ? null : poolStats.getActive());
-                table.addCell(poolStats == null ? null : poolStats.getThreads());
-                table.addCell(poolStats == null ? null : poolStats.getQueue());
+                table.addCell(active);
+                table.addCell(threads);
+                table.addCell(queue);
                 table.addCell(maxQueueSize == null ? -1 : maxQueueSize);
-                table.addCell(poolStats == null ? null : poolStats.getRejected());
-                table.addCell(poolStats == null ? null : poolStats.getLargest());
-                table.addCell(poolStats == null ? null : poolStats.getCompleted());
-                table.addCell(poolStats == null ? null : poolStats.getWaitTime());
+                table.addCell(rejected);
+                table.addCell(largest);
+                table.addCell(completed);
+                table.addCell(waitTime);
                 table.addCell(core);
                 table.addCell(max);
                 table.addCell(size);

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

@@ -469,6 +469,10 @@ private void validateSetting(Settings tpSettings) {
             }
             Settings tpGroup = entry.getValue();
             ExecutorHolder holder = executors.get(tpName);
+            // Skip validation for ForkJoinPool type since it does not support these settings
+            if (holder.info.type == ThreadPoolType.FORK_JOIN) {
+                continue;
+            }
             assert holder.executor instanceof OpenSearchThreadPoolExecutor;
             OpenSearchThreadPoolExecutor threadPoolExecutor = (OpenSearchThreadPoolExecutor) holder.executor;
             if (holder.info.type == ThreadPoolType.SCALING) {
@@ -504,6 +508,9 @@ public void setThreadPool(Settings tpSettings) {
             String tpName = entry.getKey();
             Settings tpGroup = entry.getValue();
             ExecutorHolder holder = executors.get(tpName);
+            if (holder.info.type == ThreadPoolType.FORK_JOIN) {
+                continue;
+            }
             assert holder.executor instanceof OpenSearchThreadPoolExecutor;
             OpenSearchThreadPoolExecutor executor = (OpenSearchThreadPoolExecutor) holder.executor;
             if (holder.info.type == ThreadPoolType.SCALING) {
@@ -543,6 +550,11 @@ public ThreadPoolStats stats() {
             if ("same".equals(name)) {
                 continue;
             }
+            if (holder.info.type == ThreadPoolType.FORK_JOIN) {
+                // Add ForkJoinPool with sentinel values
+                stats.add(new ThreadPoolStats.Stats(name, 0, 0, 0, 0, 0, 0, -1));
+                continue;
+            }
             int threads = -1;
             int queue = -1;
             int active = -1;
@@ -664,8 +676,11 @@ public void shutdown() {
         stopCachedTimeThread();
         scheduler.shutdown();
         for (ExecutorHolder executor : executors.values()) {
-            if (executor.executor() instanceof ThreadPoolExecutor || executor.executor() instanceof ForkJoinPool) {
-                executor.executor().shutdown();
+            ExecutorService es = executor.executor();
+            if (es instanceof ThreadPoolExecutor) {
+                es.shutdown();
+            } else if (es instanceof ForkJoinPool) {
+                es.shutdown();
             }
         }
     }
@@ -674,23 +689,50 @@ public void shutdownNow() {
         stopCachedTimeThread();
         scheduler.shutdownNow();
         for (ExecutorHolder executor : executors.values()) {
-            if (executor.executor() instanceof ThreadPoolExecutor || executor.executor() instanceof ForkJoinPool) {
-                executor.executor().shutdownNow();
+            ExecutorService es = executor.executor();
+            if (es instanceof ThreadPoolExecutor) {
+                es.shutdownNow();
+            } else if (es instanceof ForkJoinPool) {
+                es.shutdownNow(); // same as shutdown(), but explicit to the reader
             }
         }
     }
 
     public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
-        boolean result = scheduler.awaitTermination(timeout, unit);
+        long nanos = unit.toNanos(timeout);
+        long deadline = System.nanoTime() + nanos;
+        boolean result = scheduler.awaitTermination(Math.max(0, nanos), TimeUnit.NANOSECONDS);
+        long now = System.nanoTime();
+        nanos = deadline - now;
+
         for (ExecutorHolder executor : executors.values()) {
             if (executor.executor() instanceof ThreadPoolExecutor || executor.executor() instanceof ForkJoinPool) {
-                result &= executor.executor().awaitTermination(timeout, unit);
+                if (nanos <= 0) {
+                    result = false;
+                    break;
+                }
+                result &= executor.executor().awaitTermination(Math.max(0, nanos), TimeUnit.NANOSECONDS);
+                now = System.nanoTime();
+                nanos = deadline - now;
             }
         }
-        cachedTimeThread.join(unit.toMillis(timeout));
+        if (nanos > 0) {
+            cachedTimeThread.join(TimeUnit.NANOSECONDS.toMillis(nanos));
+        }
         return result;
     }
 
+    // public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+    // boolean result = scheduler.awaitTermination(timeout, unit);
+    // for (ExecutorHolder executor : executors.values()) {
+    // if (executor.executor() instanceof ThreadPoolExecutor || executor.executor() instanceof ForkJoinPool) {
+    // result &= executor.executor().awaitTermination(timeout, unit);
+    // }
+    // }
+    // cachedTimeThread.join(unit.toMillis(timeout));
+    // return result;
+    // }
+
     public ScheduledExecutorService scheduler() {
         return this.scheduler;
     }
@@ -948,6 +990,10 @@ public void writeTo(StreamOutput out) throws IOException {
                 // Opensearch on older version doesn't know about "resizable" thread pool. Convert RESIZABLE to FIXED
                 // to avoid serialization/de-serization issue between nodes with different OpenSearch version
                 out.writeString(ThreadPoolType.FIXED.getType());
+            } else if (type == ThreadPoolType.FORK_JOIN && out.getVersion().before(Version.V_3_2_0)) {
+                // Opensearch on older version doesn't know about "fork_join" thread pool. Convert FORK_JOIN to FIXED (or SCALING, or
+                // another safe fallback)
+                out.writeString(ThreadPoolType.FIXED.getType());
             } else {
                 out.writeString(type.getType());
             }
@@ -993,6 +1039,12 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                 builder.field("core", min);
                 assert max != -1;
                 builder.field("max", max);
+            } else if (type == ThreadPoolType.FORK_JOIN) {
+                builder.field("size", -1);
+                builder.field("min", -1);
+                builder.field("max", -1);
+                builder.field("keep_alive", (Object) null);
+                builder.field("queue_size", -1);
             } else {
                 assert max != -1;
                 builder.field("size", max);