Fix exception cause logging in QueryResultPusher (#14975)

processing/src/main/java/org/apache/druid/error/QueryExceptionCompat.java

@@ -47,7 +47,7 @@ public class QueryExceptionCompat extends DruidException.Failure
   {
     return bob.forPersona(DruidException.Persona.OPERATOR)
               .ofCategory(convertFailType(exception.getFailType()))
-              .build(exception.getMessage())
+              .build(exception, exception.getMessage())
               .withContext("host", exception.getHost())
               .withContext("errorClass", exception.getErrorClass())
               .withContext("legacyErrorCode", exception.getErrorCode());

server/src/test/java/org/apache/druid/server/QueryResourceTest.java

@@ -23,6 +23,7 @@ import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.jaxrs.smile.SmileMediaTypes;
 import com.google.common.base.Suppliers;
+import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -87,6 +88,7 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 import javax.servlet.http.HttpServletResponse;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
@@ -375,6 +377,87 @@ public class QueryResourceTest
     );
   }
 
+
+  @Test
+  public void testQueryThrowsRuntimeExceptionFromLifecycleExecute() throws IOException
+  {
+    String embeddedExceptionMessage = "Embedded Exception Message!";
+    String overrideConfigKey = "priority";
+    String overrideConfigValue = "678";
+
+    DefaultQueryConfig overrideConfig = new DefaultQueryConfig(ImmutableMap.of(overrideConfigKey, overrideConfigValue));
+    QuerySegmentWalker querySegmentWalker = new QuerySegmentWalker()
+    {
+      @Override
+      public <T> QueryRunner<T> getQueryRunnerForIntervals(
+          Query<T> query,
+          Iterable<Interval> intervals
+      )
+      {
+        throw new RuntimeException("something", new RuntimeException(embeddedExceptionMessage));
+      }
+
+      @Override
+      public <T> QueryRunner<T> getQueryRunnerForSegments(
+          Query<T> query,
+          Iterable<SegmentDescriptor> specs
+      )
+      {
+        throw new UnsupportedOperationException();
+      }
+    };
+
+    queryResource = new QueryResource(
+
+        new QueryLifecycleFactory(null, null, null, null, null, null, null, Suppliers.ofInstance(overrideConfig))
+        {
+          @Override
+          public QueryLifecycle factorize()
+          {
+            return new QueryLifecycle(
+                WAREHOUSE,
+                querySegmentWalker,
+                new DefaultGenericQueryMetricsFactory(),
+                new NoopServiceEmitter(),
+                testRequestLogger,
+                AuthTestUtils.TEST_AUTHORIZER_MAPPER,
+                overrideConfig,
+                new AuthConfig(),
+                System.currentTimeMillis(),
+                System.nanoTime())
+            {
+              @Override
+              public void emitLogsAndMetrics(@Nullable Throwable e, @Nullable String remoteAddress, long bytesWritten)
+              {
+                Assert.assertTrue(Throwables.getStackTraceAsString(e).contains(embeddedExceptionMessage));
+              }
+            };
+          }
+        },
+        jsonMapper,
+        smileMapper,
+        queryScheduler,
+        new AuthConfig(),
+        null,
+        ResponseContextConfig.newConfig(true),
+        DRUID_NODE
+    );
+
+    expectPermissiveHappyPathAuth();
+
+    final Response response = expectSynchronousRequestFlow(SIMPLE_TIMESERIES_QUERY);
+    Assert.assertEquals(Status.INTERNAL_SERVER_ERROR.getStatusCode(), response.getStatus());
+
+    final ErrorResponse entity = (ErrorResponse) response.getEntity();
+    MatcherAssert.assertThat(
+        entity.getUnderlyingException(),
+        new DruidExceptionMatcher(
+            DruidException.Persona.OPERATOR,
+            DruidException.Category.RUNTIME_FAILURE, "legacyQueryException")
+            .expectMessageIs("something")
+    );
+  }
+
   @Test
   public void testGoodQueryWithQueryConfigDoesNotOverrideQueryContext() throws IOException
   {

server/src/test/java/org/apache/druid/server/QueryResultPusherTest.java

@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Throwables;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.server.QueryResource.QueryMetricCounter;
+import org.apache.druid.server.QueryResultPusher.ResultsWriter;
+import org.apache.druid.server.QueryResultPusher.Writer;
+import org.apache.druid.server.mocks.MockHttpServletRequest;
+import org.junit.Test;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response.ResponseBuilder;
+
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.Assert.assertTrue;
+
+public class QueryResultPusherTest
+{
+  private static final DruidNode DRUID_NODE = new DruidNode(
+      "broker",
+      "localhost",
+      true,
+      8082,
+      null,
+      true,
+      false);
+
+  @Test
+  public void testResultPusherRetainsNestedExceptionBacktraces()
+  {
+
+    HttpServletRequest request = new MockHttpServletRequest();
+    ObjectMapper jsonMapper = new DefaultObjectMapper();
+    ResponseContextConfig responseContextConfig = ResponseContextConfig.newConfig(true);
+    DruidNode selfNode = DRUID_NODE;
+    QueryResource.QueryMetricCounter counter = new NoopQueryMetricCounter();
+    String queryId = "someQuery";
+    MediaType contentType = MediaType.APPLICATION_JSON_TYPE;
+    Map<String, String> extraHeaders = new HashMap<String, String>();
+    AtomicBoolean recordFailureInvoked = new AtomicBoolean();
+
+    String embeddedExceptionMessage = "Embedded Exception Message!";
+    RuntimeException embeddedException = new RuntimeException(embeddedExceptionMessage);
+    RuntimeException topException = new RuntimeException("Where's the party?", embeddedException);
+
+    ResultsWriter resultWriter = new ResultsWriter()
+    {
+
+      @Override
+      public void close()
+      {
+      }
+
+      @Override
+      public ResponseBuilder start()
+      {
+        throw topException;
+      }
+
+      @Override
+      public void recordSuccess(long numBytes)
+      {
+      }
+
+      @Override
+      public void recordFailure(Exception e)
+      {
+        assertTrue(Throwables.getStackTraceAsString(e).contains(embeddedExceptionMessage));
+        recordFailureInvoked.set(true);
+      }
+
+      @Override
+      public Writer makeWriter(OutputStream out)
+      {
+        return null;
+      }
+
+      @Override
+      public QueryResponse<Object> getQueryResponse()
+      {
+        return null;
+      }
+    };
+    QueryResultPusher pusher = new QueryResultPusher(
+        request,
+        jsonMapper,
+        responseContextConfig,
+        selfNode,
+        counter,
+        queryId,
+        contentType,
+        extraHeaders)
+    {
+
+      @Override
+      public void writeException(Exception e, OutputStream out)
+      {
+      }
+
+      @Override
+      public ResultsWriter start()
+      {
+        return resultWriter;
+      }
+    };
+
+    pusher.push();
+
+    assertTrue("recordFailure(e) should have been invoked!", recordFailureInvoked.get());
+  }
+
+  static class NoopQueryMetricCounter implements QueryMetricCounter
+  {
+
+    @Override
+    public void incrementSuccess()
+    {
+    }
+
+    @Override
+    public void incrementFailed()
+    {
+    }
+
+    @Override
+    public void incrementInterrupted()
+    {
+    }
+
+    @Override
+    public void incrementTimedOut()
+    {
+    }
+
+  }
+}