Avoid losing error message in failure collector (#111983)

The node-disconnected exception might not include the root cause. In 
this case, the failure collector incorrectly unwraps the exception and
wraps it in a new Elasticsearch exception, losing the message. We should
instead use the original exception to preserve the reason.

Closes #111894

Author: dnhatn

docs/changelog/111983.yaml

@@ -0,0 +1,6 @@
+pr: 111983
+summary: Avoid losing error message in failure collector
+area: ES|QL
+type: bug
+issues:
+ - 111894

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

@@ -46,10 +46,19 @@ public FailureCollector(int maxExceptions) {
         this.maxExceptions = maxExceptions;
     }
 
-    public void unwrapAndCollect(Exception originEx) {
-        final Exception e = originEx instanceof TransportException
-            ? (originEx.getCause() instanceof Exception cause ? cause : new ElasticsearchException(originEx.getCause()))
-            : originEx;
+    private static Exception unwrapTransportException(TransportException te) {
+        final Throwable cause = te.getCause();
+        if (cause == null) {
+            return te;
+        } else if (cause instanceof Exception ex) {
+            return ex;
+        } else {
+            return new ElasticsearchException(cause);
+        }
+    }
+
+    public void unwrapAndCollect(Exception e) {
+        e = e instanceof TransportException te ? unwrapTransportException(te) : e;
         if (ExceptionsHelper.unwrap(e, TaskCancelledException.class) != null) {
             if (cancelledExceptionsCount.incrementAndGet() <= maxExceptions) {
                 cancelledExceptions.add(e);

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/FailureCollectorTests.java

@@ -7,12 +7,15 @@
 
 package org.elasticsearch.compute.operator;
 
+import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.breaker.CircuitBreakingException;
 import org.elasticsearch.tasks.TaskCancelledException;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.NodeDisconnectedException;
 import org.elasticsearch.transport.RemoteTransportException;
+import org.elasticsearch.transport.TransportException;
 import org.hamcrest.Matchers;
 
 import java.io.IOException;
@@ -25,6 +28,9 @@
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
+import static org.hamcrest.Matchers.arrayWithSize;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.lessThan;
 
 public class FailureCollectorTests extends ESTestCase {
@@ -87,4 +93,17 @@ public void testEmpty() {
         assertFalse(collector.hasFailure());
         assertNull(collector.getFailure());
     }
+
+    public void testTransportExceptions() {
+        FailureCollector collector = new FailureCollector(5);
+        collector.unwrapAndCollect(new NodeDisconnectedException(DiscoveryNodeUtils.builder("node-1").build(), "/field_caps"));
+        collector.unwrapAndCollect(new TransportException(new CircuitBreakingException("too large", CircuitBreaker.Durability.TRANSIENT)));
+        Exception failure = collector.getFailure();
+        assertNotNull(failure);
+        assertThat(failure, instanceOf(NodeDisconnectedException.class));
+        assertThat(failure.getMessage(), equalTo("[][0.0.0.0:1][/field_caps] disconnected"));
+        Throwable[] suppressed = failure.getSuppressed();
+        assertThat(suppressed, arrayWithSize(1));
+        assertThat(suppressed[0], instanceOf(CircuitBreakingException.class));
+    }
 }