Refactor the listener code

Author: smalyshev

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

@@ -92,7 +92,12 @@ void startComputeOnRemoteCluster(
                 childSessionId,
                 queryPragmas.exchangeBufferSize(),
                 esqlExecutor,
-                ActionListener.wrap(unused -> {
+                EsqlCCSUtils.skipUnavailableListener(
+                    openExchangeListener,
+                    executionInfo,
+                    clusterAlias,
+                    EsqlExecutionInfo.Cluster.Status.SKIPPED
+                ).delegateFailureAndWrap((l, unused) -> {
                     var listenerGroup = new RemoteListenerGroup(
                         transportService,
                         rootTask,
@@ -129,18 +134,6 @@ void startComputeOnRemoteCluster(
                         TransportRequestOptions.EMPTY,
                         new ActionListenerResponseHandler<>(clusterListener, ComputeResponse::new, esqlExecutor)
                     );
-                }, e -> {
-                    if (EsqlCCSUtils.shouldIgnoreRuntimeError(executionInfo, clusterAlias, e)) {
-                        EsqlCCSUtils.markClusterWithFinalStateAndNoShards(
-                            executionInfo,
-                            clusterAlias,
-                            EsqlExecutionInfo.Cluster.Status.SKIPPED,
-                            e
-                        );
-                        openExchangeListener.onResponse(null);
-                    } else {
-                        openExchangeListener.onFailure(e);
-                    }
                 })
             );
         }

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

@@ -74,17 +74,8 @@ 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(() -> {
-                if (EsqlCCSUtils.shouldIgnoreRuntimeError(executionInfo, clusterAlias, e)) {
-                    EsqlCCSUtils.markClusterWithFinalStateAndNoShards(
-                        executionInfo,
-                        clusterAlias,
-                        EsqlExecutionInfo.Cluster.Status.PARTIAL,
-                        e
-                    );
-                    delegate.onResponse(null);
-                } else {
-                    delegate.onFailure(e);
-                }
+                EsqlCCSUtils.skipUnavailableListener(delegate, executionInfo, clusterAlias, EsqlExecutionInfo.Cluster.Status.PARTIAL)
+                    .onFailure(e);
             }));
         }), finishGroup);
     }

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

@@ -369,4 +369,23 @@ public static boolean shouldIgnoreRuntimeError(EsqlExecutionInfo executionInfo,
 
         return ExceptionsHelper.isRemoteUnavailableException(e);
     }
+
+    /**
+     * Wrap a listener so that it will skip errors that are ignorable
+     */
+    public static <T> ActionListener<T> skipUnavailableListener(
+        ActionListener<T> delegate,
+        EsqlExecutionInfo executionInfo,
+        String clusterAlias,
+        EsqlExecutionInfo.Cluster.Status status
+    ) {
+        return delegate.delegateResponse((l, e) -> {
+            if (EsqlCCSUtils.shouldIgnoreRuntimeError(executionInfo, clusterAlias, e)) {
+                EsqlCCSUtils.markClusterWithFinalStateAndNoShards(executionInfo, clusterAlias, status, e);
+                l.onResponse(null);
+            } else {
+                l.onFailure(e);
+            }
+        });
+    }
 }