Update ExecutorScalingQueue to workaround LinkedTransferQueue JDK bug

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

@@ -329,6 +329,28 @@ public boolean offer(E e) {
             }
         }
 
+        // The following three methods, put, add, and timed offer, are
+        // 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) {
+            boolean added = super.offer(e);
+            assert added;
+        }
+
+        @Override
+        public boolean add(E e) {
+            boolean added = super.offer(e);
+            assert added;
+            return added;
+        }
+
+        @Override
+        public boolean offer(E e, long timeout, TimeUnit unit) {
+            boolean added = super.offer(e);
+            assert added;
+            return added;
+        }
     }
 
     /**

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

@@ -12,6 +12,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.Processors;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.hamcrest.Matcher;
 
 import java.util.Locale;
@@ -501,4 +502,41 @@ public void testNodeProcessorsFloatValidation() {
         }
     }
 
+    // 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 testLTQIssue() {
+        final var executorService = EsExecutors.newScaling(
+            "test",
+            0,
+            1,
+            60,
+            TimeUnit.SECONDS,
+            true,
+            EsExecutors.daemonThreadFactory("test"),
+            new ThreadContext(Settings.EMPTY)
+        );
+        try {
+            final var countDownLatch = new CountDownLatch(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);
+                }
+            }
+
+            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,35 @@
+/*
+ * 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.TimeUnit;
+
+public class ExecutorScalingQueueTests extends ESTestCase {
+
+    public void testPut() {
+        var queue = new EsExecutors.ExecutorScalingQueue<>();
+        queue.put(new Object());
+        assertEquals(queue.size(), 1);
+    }
+
+    public void testAdd() {
+        var queue = new EsExecutors.ExecutorScalingQueue<>();
+        queue.add(new Object());
+        assertEquals(queue.size(), 1);
+    }
+
+    public void testTimedOffer() {
+        var queue = new EsExecutors.ExecutorScalingQueue<>();
+        queue.offer(new Object(), 60, TimeUnit.SECONDS);
+        assertEquals(queue.size(), 1);
+    }
+
+}