From c5514c1d9484dd0cae364dcb63ad4c5ea5a973b7 Mon Sep 17 00:00:00 2001
From: Tommy Ludwig <8924140+shakuzen@users.noreply.github.com>
Date: Thu, 5 Dec 2024 18:39:47 +0900
Subject: [PATCH] Include queued submissions in `executor.queued`

This was missed when the metric was originally implemented for ForkJoinPool. Also aligns the description with the other `executor.queued` gauge for `ThreadPoolExecutor`.

Resolves gh-5650
---
 .../binder/jvm/ExecutorServiceMetrics.java    |  6 +++--
 .../jvm/ExecutorServiceMetricsTest.java       | 27 +++++++++++++++++++
 2 files changed, 31 insertions(+), 2 deletions(-)

diff --git a/micrometer-core/src/main/java/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetrics.java b/micrometer-core/src/main/java/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetrics.java
index f0bbf7038c..9f01c06a8a 100644
--- a/micrometer-core/src/main/java/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetrics.java
+++ b/micrometer-core/src/main/java/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetrics.java
@@ -387,9 +387,11 @@ private void monitor(MeterRegistry registry, ForkJoinPool fj) {
                     + "underestimates the actual total number of steals when the pool " + "is not quiescent")
             .register(registry);
 
-        Gauge.builder(metricPrefix + "executor.queued", fj, ForkJoinPool::getQueuedTaskCount)
+        Gauge
+            .builder(metricPrefix + "executor.queued", fj,
+                    pool -> pool.getQueuedTaskCount() + pool.getQueuedSubmissionCount())
             .tags(tags)
-            .description("An estimate of the total number of tasks currently held in queues by worker threads")
+            .description("The approximate number of tasks that are queued for execution")
             .register(registry);
 
         Gauge.builder(metricPrefix + "executor.active", fj, ForkJoinPool::getActiveThreadCount)
diff --git a/micrometer-core/src/test/java/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetricsTest.java b/micrometer-core/src/test/java/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetricsTest.java
index 296452723f..be3d1d9d3f 100644
--- a/micrometer-core/src/test/java/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetricsTest.java
+++ b/micrometer-core/src/test/java/io/micrometer/core/instrument/binder/jvm/ExecutorServiceMetricsTest.java
@@ -31,6 +31,7 @@
 import org.junit.jupiter.params.provider.CsvSource;
 
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.assertj.core.api.AssertionsForClassTypes.*;
 
@@ -259,6 +260,32 @@ void monitorScheduledExecutorServiceWithRepetitiveTasks(String metricPrefix, Str
         assertThat(registry.get(expectedMetricPrefix + "executor.idle").tags(userTags).timer().count()).isEqualTo(0L);
     }
 
+    @Test
+    @Issue("#5650")
+    void queuedSubmissionsAreIncludedInExecutorQueuedMetric() {
+        ForkJoinPool pool = new ForkJoinPool(1, ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, false, 1, 1, 1,
+                a -> true, 555, TimeUnit.MILLISECONDS);
+        ExecutorServiceMetrics.monitor(registry, pool, "myForkJoinPool");
+        AtomicBoolean busy = new AtomicBoolean(true);
+
+        // will be an active task
+        pool.execute(() -> {
+            while (busy.get()) {
+            }
+        });
+
+        // will be queued for submission
+        pool.execute(() -> {
+        });
+        pool.execute(() -> {
+        });
+
+        double queued = registry.get("executor.queued").tag("name", "myForkJoinPool").gauge().value();
+        busy.set(false);
+
+        assertThat(queued).isEqualTo(2.0);
+    }
+
     @SuppressWarnings("unchecked")
     private <T extends Executor> T monitorExecutorService(String executorName, String metricPrefix, T exec) {
         if (metricPrefix == null) {