[7.17] Update ExecutorScalingQueue to workaround LinkedTransferQueue JDK bug (#104347) (#104425)

* Update ExecutorScalingQueue to workaround LinkedTransferQueue JDK bug (#104347)

This commit adds a few overrides to ExecutorScalingQueue (subclass of LinkedTransferQueue) to workaround a JDK bug in LinkedTransferQueue.

* remove usage of var

* fix compilation

Author: ChrisHegarty

server/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java

@@ -382,6 +382,29 @@ public boolean offer(E e) {
             }
         }
 
+        // Overridden to workaround a JDK bug introduced in JDK 21.0.2
+        // https://bugs.openjdk.org/browse/JDK-8323659
+        @Override
+        public void put(E e) {
+            // As the queue is unbounded, this method will always add to the queue.
+            super.offer(e);
+        }
+
+        // Overridden to workaround a JDK bug introduced in JDK 21.0.2
+        // https://bugs.openjdk.org/browse/JDK-8323659
+        @Override
+        public boolean add(E e) {
+            // As the queue is unbounded, this method will never return false.
+            return super.offer(e);
+        }
+
+        // Overridden to workaround a JDK bug introduced in JDK 21.0.2
+        // https://bugs.openjdk.org/browse/JDK-8323659
+        @Override
+        public boolean offer(E e, long timeout, TimeUnit unit) {
+            // As the queue is unbounded, this method will never return false.
+            return super.offer(e);
+        }
     }
 
     /**

server/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.hamcrest.Matcher;
 
 import java.util.Locale;
@@ -437,4 +438,41 @@ private void runProcessorsBoundTest(final Setting<Integer> processorsSetting) {
         assertSettingDeprecationsAndWarnings(deprecatedSettings, new DeprecationWarning(DeprecationLogger.CRITICAL, expectedWarning));
     }
 
+    // This test must complete to ensure that our basic infrastructure is working as expected.
+    // Specifically that ExecutorScalingQueue, which subclasses LinkedTransferQueue, correctly
+    // tracks tasks submitted to the executor.
+    public void testBasicTaskExecution() {
+        final ThreadPoolExecutor executorService = EsExecutors.newScaling(
+            "test",
+            0,
+            between(1, 5),
+            60,
+            TimeUnit.SECONDS,
+            randomBoolean(),
+            EsExecutors.daemonThreadFactory("test"),
+            new ThreadContext(Settings.EMPTY)
+        );
+        try {
+            final CountDownLatch countDownLatch = new CountDownLatch(between(1, 10));
+            class TestTask extends AbstractRunnable {
+                @Override
+                protected void doRun() {
+                    countDownLatch.countDown();
+                    if (countDownLatch.getCount() > 0) {
+                        executorService.execute(TestTask.this);
+                    }
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    fail(e.getMessage());
+                }
+            }
+
+            executorService.execute(new TestTask());
+            safeAwait(countDownLatch);
+        } finally {
+            ThreadPool.terminate(executorService, 10, TimeUnit.SECONDS);
+        }
+    }
 }

server/src/test/java/org/elasticsearch/common/util/concurrent/ExecutorScalingQueueTests.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.common.util.concurrent;
+
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.concurrent.LinkedTransferQueue;
+import java.util.concurrent.TimeUnit;
+
+public class ExecutorScalingQueueTests extends ESTestCase {
+
+    public void testPut() {
+        LinkedTransferQueue<Object> queue = new EsExecutors.ExecutorScalingQueue<>();
+        queue.put(new Object());
+        assertEquals(queue.size(), 1);
+    }
+
+    public void testAdd() {
+        LinkedTransferQueue<Object> queue = new EsExecutors.ExecutorScalingQueue<>();
+        assertTrue(queue.add(new Object()));
+        assertEquals(queue.size(), 1);
+    }
+
+    public void testTimedOffer() {
+        LinkedTransferQueue<Object> queue = new EsExecutors.ExecutorScalingQueue<>();
+        assertTrue(queue.offer(new Object(), 60, TimeUnit.SECONDS));
+        assertEquals(queue.size(), 1);
+    }
+
+}