Fix listener leak in exchange service (elastic#122417) (elastic#122454) (elastic#122467)

If we hit the circuit breaker exception before fetching pages,
we fail to notify the listener.

Closes elastic#122271

Author: dnhatn

docs/changelog/122417.yaml

@@ -0,0 +1,6 @@
+pr: 122417
+summary: Fix listener leak in exchange service
+area: ES|QL
+type: bug
+issues:
+ - 122271

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/EsqlRefCountingListener.java

@@ -34,7 +34,8 @@ public EsqlRefCountingListener(ActionListener<Void> delegate) {
     }
 
     public ActionListener<Void> acquire() {
-        return refs.acquireListener().delegateResponse((l, e) -> {
+        var listener = ActionListener.assertAtLeastOnce(refs.acquireListener());
+        return listener.delegateResponse((l, e) -> {
             failureCollector.unwrapAndCollect(e);
             l.onFailure(e);
         });

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.support.ChannelActionListener;
+import org.elasticsearch.common.breaker.CircuitBreakingException;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -334,7 +335,13 @@ private void doFetchPageAsync(boolean allSourcesFinished, ActionListener<Exchang
             final long reservedBytes = allSourcesFinished ? 0 : estimatedPageSizeInBytes.get();
             if (reservedBytes > 0) {
                 // This doesn't fully protect ESQL from OOM, but reduces the likelihood.
-                blockFactory.breaker().addEstimateBytesAndMaybeBreak(reservedBytes, "fetch page");
+                try {
+                    blockFactory.breaker().addEstimateBytesAndMaybeBreak(reservedBytes, "fetch page");
+                } catch (Exception e) {
+                    assert e instanceof CircuitBreakingException : new AssertionError(e);
+                    listener.onFailure(e);
+                    return;
+                }
                 listener = ActionListener.runAfter(listener, () -> blockFactory.breaker().addWithoutBreaking(-reservedBytes));
             }
             transportService.sendChildRequest(

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeListener.java

@@ -174,7 +174,8 @@ private boolean isCCSListener(String computeClusterAlias) {
      * Acquires a new listener that doesn't collect result
      */
     ActionListener<Void> acquireAvoid() {
-        return refs.acquire().delegateResponse((l, e) -> {
+        var listener = ActionListener.assertAtLeastOnce(refs.acquire());
+        return listener.delegateResponse((l, e) -> {
             try {
                 if (cancelled.compareAndSet(false, true)) {
                     LOGGER.debug("cancelling ESQL task {} on failure", task);