Fix race in AbstractSearchAsyncAction request throttling (#116264)

We had a race here where the non-blocking pending execution
would be starved of executing threads.
This happened when all the current holders of permits from the semaphore
would release their permit after a producer thread failed to acquire a
permit and then enqueued its task.
=> need to peek the queue again after releasing the permit and try to
acquire a new permit if there's work left to be done to avoid this
scenario.

Author: original-brownbear

muted-tests.yml

@@ -237,9 +237,6 @@ tests:
 - class: org.elasticsearch.search.basic.SearchWithRandomDisconnectsIT
   method: testSearchWithRandomDisconnects
   issue: https://github.com/elastic/elasticsearch/issues/116175
-- class: org.elasticsearch.indexing.IndexActionIT
-  method: testAutoGenerateIdNoDuplicates
-  issue: https://github.com/elastic/elasticsearch/issues/115716
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=ml/start_stop_datafeed/Test start datafeed given index pattern with no matching indices}
   issue: https://github.com/elastic/elasticsearch/issues/116220

server/src/main/java/org/elasticsearch/action/search/AbstractSearchAsyncAction.java

@@ -48,8 +48,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
-import java.util.concurrent.LinkedTransferQueue;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -751,7 +751,7 @@ protected final ShardSearchRequest buildShardSearchRequest(SearchShardIterator s
 
     private static final class PendingExecutions {
         private final Semaphore semaphore;
-        private final LinkedTransferQueue<Consumer<Releasable>> queue = new LinkedTransferQueue<>();
+        private final ConcurrentLinkedQueue<Consumer<Releasable>> queue = new ConcurrentLinkedQueue<>();
 
         PendingExecutions(int permits) {
             assert permits > 0 : "not enough permits: " + permits;
@@ -770,11 +770,10 @@ void submit(Consumer<Releasable> task) {
                     }
                 }
             }
-
         }
 
         private void executeAndRelease(Consumer<Releasable> task) {
-            while (task != null) {
+            do {
                 final SubscribableListener<Void> onDone = new SubscribableListener<>();
                 task.accept(() -> onDone.onResponse(null));
                 if (onDone.isDone()) {
@@ -797,13 +796,21 @@ public void onFailure(Exception e) {
                     });
                     return;
                 }
-            }
+            } while (task != null);
         }
 
         private Consumer<Releasable> pollNextTaskOrReleasePermit() {
             var task = queue.poll();
             if (task == null) {
                 semaphore.release();
+                while (queue.peek() != null && semaphore.tryAcquire()) {
+                    task = queue.poll();
+                    if (task == null) {
+                        semaphore.release();
+                    } else {
+                        return task;
+                    }
+                }
             }
             return task;
         }