Fix listener usage in remote clusters (#122629) (#122647)

We should notify the listener provided in delegateResponse/delegateFailure, 
not the original listener.

Author: dnhatn

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

@@ -85,15 +85,14 @@ void startComputeOnRemoteCluster(
             var resp = finalResponse.get();
             return Objects.requireNonNullElseGet(resp, () -> new ComputeResponse(profiles));
         }))) {
-            var openExchangeListener = computeListener.acquireAvoid();
             ExchangeService.openExchange(
                 transportService,
                 cluster.connection,
                 childSessionId,
                 queryPragmas.exchangeBufferSize(),
                 esqlExecutor,
                 EsqlCCSUtils.skipUnavailableListener(
-                    openExchangeListener,
+                    computeListener.acquireAvoid(),
                     executionInfo,
                     clusterAlias,
                     EsqlExecutionInfo.Cluster.Status.SKIPPED
@@ -104,7 +103,7 @@ void startComputeOnRemoteCluster(
                         computeListener,
                         clusterAlias,
                         executionInfo,
-                        openExchangeListener
+                        l
                     );
 
                     var remoteSink = exchangeService.newRemoteSink(

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

@@ -74,7 +74,7 @@ class RemoteListenerGroup {
     private <T> ActionListener<T> createCancellingListener(String reason, ActionListener<T> delegate, Runnable finishGroup) {
         return ActionListener.runAfter(delegate.delegateResponse((inner, e) -> {
             taskManager.cancelTaskAndDescendants(groupTask, reason, true, ActionListener.running(() -> {
-                EsqlCCSUtils.skipUnavailableListener(delegate, executionInfo, clusterAlias, EsqlExecutionInfo.Cluster.Status.PARTIAL)
+                EsqlCCSUtils.skipUnavailableListener(inner, executionInfo, clusterAlias, EsqlExecutionInfo.Cluster.Status.PARTIAL)
                     .onFailure(e);
             }));
         }), finishGroup);