Fixing a race condition in EnrichCoordinatorProxyAction that can leave an item stuck in its queue (#90688)

There is a race condition in EnrichCoordinatorProxyAction that can result in an item being stuck in its
queue even once all threads related to any schedule() calls have completed. The item will be flushed
out on the next call to schedule() but there is no guarantee if or when that will happen. This PR adds
an additional check for orphaned items in the queue.

docs/changelog/90688.yaml

@@ -0,0 +1,7 @@
+pr: 90688
+summary: Fixing a race condition in `EnrichCoordinatorProxyAction` that can leave
+  an item stuck in its queue
+area: Ingest Node
+type: bug
+issues:
+ - 90598

x-pack/plugin/enrich/src/main/java/org/elasticsearch/xpack/enrich/action/EnrichCoordinatorProxyAction.java

@@ -160,7 +160,17 @@ public class EnrichCoordinatorProxyAction extends ActionType<SearchResponse> {
                 final List<Slot> slots = new ArrayList<>(Math.min(queue.size(), maxLookupsPerRequest));
                 if (queue.drainTo(slots, maxLookupsPerRequest) == 0) {
                     remoteRequestPermits.release();
-                    return;
+                    /*
+                     * It is possible that something was added to the queue after the drain and before the permit was released, meaning
+                     * that the other thread could not acquire the permit, leaving an item orphaned in the queue. So we check the queue
+                     * again after releasing the permit, and if there is something there we run another loop to pick that thing up. If
+                     * another thread has picked it up in the meantime, we'll just exit out of the loop on the next try.
+                     */
+                    if (queue.isEmpty()) {
+                        return;
+                    } else {
+                        continue;
+                    }
                 }
                 assert slots.isEmpty() == false;
                 remoteRequestsTotal.increment();