Add test and handle missed subset of scenarios where an error can be

thrown

Author: pawankartik-elastic

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterRemoteErrorClusterAliasWrapIT.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.action;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.compute.operator.exchange.ExchangeService;
+import org.elasticsearch.test.transport.MockTransportService;
+import org.elasticsearch.transport.TransportService;
+
+import static org.hamcrest.Matchers.is;
+
+public class CrossClusterRemoteErrorClusterAliasWrapIT extends AbstractCrossClusterTestCase {
+
+    public void testThatRemoteErrorsAreWrapped() throws Exception {
+        setupClusters(2);
+        setSkipUnavailable(REMOTE_CLUSTER_1, false);
+        setSkipUnavailable(REMOTE_CLUSTER_2, false);
+
+        /*
+         * Let's say something went wrong with the Exchange and its specifics when talking to a remote.
+         * And let's pretend only cluster-a is affected.
+         */
+        for (var nodes : cluster(REMOTE_CLUSTER_1).getNodeNames()) {
+            ((MockTransportService) cluster(REMOTE_CLUSTER_1).getInstance(TransportService.class, nodes)).addRequestHandlingBehavior(
+                ExchangeService.OPEN_EXCHANGE_ACTION_NAME,
+                (requestHandler, transportRequest, transportChannel, transportTask) -> {
+                    throw new IllegalArgumentException("some error to wreck havoc");
+                }
+            );
+        }
+
+        ElasticsearchException wrappedError = expectThrows(
+            ElasticsearchException.class,
+            () -> runQuery("FROM " + REMOTE_CLUSTER_1 + ":*," + REMOTE_CLUSTER_2 + ":* | LIMIT 100", false)
+        );
+        assertThat(wrappedError.getMessage(), is("cluster-a encountered an error"));
+    }
+}

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

@@ -199,25 +199,21 @@ record RemoteCluster(String clusterAlias, Transport.Connection connection, Strin
     public void messageReceived(ClusterComputeRequest request, TransportChannel channel, Task task) {
         ChannelActionListener<ComputeResponse> listener = new ChannelActionListener<>(channel);
         RemoteClusterPlan remoteClusterPlan = request.remoteClusterPlan();
-        try {
-            var plan = remoteClusterPlan.plan();
-            if (plan instanceof ExchangeSinkExec == false) {
-                listener.onFailure(new IllegalStateException("expected exchange sink for a remote compute; got " + plan));
-                return;
-            }
-            runComputeOnRemoteCluster(
-                request.clusterAlias(),
-                request.sessionId(),
-                (CancellableTask) task,
-                request.configuration(),
-                (ExchangeSinkExec) plan,
-                Set.of(remoteClusterPlan.targetIndices()),
-                remoteClusterPlan.originalIndices(),
-                listener
-            );
-        } catch (Exception e) {
-            listener.onFailure(new ElasticsearchException(request.clusterAlias() + " encountered an error", e));
+        var plan = remoteClusterPlan.plan();
+        if (plan instanceof ExchangeSinkExec == false) {
+            listener.onFailure(new IllegalStateException("expected exchange sink for a remote compute; got " + plan));
+            return;
         }
+        runComputeOnRemoteCluster(
+            request.clusterAlias(),
+            request.sessionId(),
+            (CancellableTask) task,
+            request.configuration(),
+            (ExchangeSinkExec) plan,
+            Set.of(remoteClusterPlan.targetIndices()),
+            remoteClusterPlan.originalIndices(),
+            listener
+        );
     }
 
     /**

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

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.plugin;
 
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.OriginalIndices;
 import org.elasticsearch.action.search.SearchRequest;
@@ -299,6 +300,9 @@ public void execute(
                         cancelQueryOnFailure,
                         execInfo,
                         computeListener.acquireCompute()
+                            .delegateResponse(
+                                (l, ex) -> l.onFailure(new ElasticsearchException(cluster.clusterAlias() + " encountered an error", ex))
+                            )
                     );
                 }
             }