Enhance json parser error logging to better track Istio Proxy error message (#15176)

Currently the inter Druid communication via rest endpoints is based on json formatted payload. Upon parsing error, there is only a generic exception stating expected json token type and current json token type. There is no detailed error log about the content of the payload causing the violation.

In the micro-service world, the trend is to deploy the Druid servers in k8 with the mesh network. Often the istio proxy or other proxies is used to intercept the network connection between Druid servers. The proxy may give error messages for various reasons. These error messages are not expected by the json parser. The generic error message from Druid can be very misleading as the user may think the message is based on the response from the other Druid server.

For example, this is an example of mysterious error message

QueryInterruptedException{msg=Next token wasn't a START_ARRAY, was[VALUE_STRING] from url[http://xxxxx:8088/druid/v2/], code=Unknown exception, class=org.apache.druid.java.util.common.IAE, host=xxxxx:8088}"

While the context of the message is the following from the proxy when it can't tunnel the network connection.

pstream connect error or disconnect/reset before header

So this very simple PR is just to enhance the logging and get the real underlying message printed out. This would save a lot of head scratching time if Druid is deployed with mesh network.

Co-authored-by: Kai Sun <kai.sun@salesforce.com>

server/src/main/java/org/apache/druid/client/JsonParserIterator.java

@@ -185,8 +185,17 @@ public class JsonParserIterator<T> implements CloseableIterator<T>
         } else if (nextToken == JsonToken.START_OBJECT) {
           throw convertException(jp.getCodec().readValue(jp, QueryException.class));
         } else {
+          String errMsg = jp.getValueAsString();
+          if (errMsg != null) {
+            errMsg = errMsg.substring(0, Math.min(errMsg.length(), 192));
+          }
           throw convertException(
-              new IAE("Next token wasn't a START_ARRAY, was[%s] from url[%s]", jp.getCurrentToken(), url)
+              new IAE(
+                  "Next token wasn't a START_ARRAY, was[%s] from url[%s] with value[%s]",
+                  jp.getCurrentToken(),
+                  url,
+                  errMsg
+              )
           );
         }
       }

server/src/test/java/org/apache/druid/client/JsonParserIteratorTest.java

@@ -316,8 +316,60 @@ public class JsonParserIteratorTest
     }
   }
 
+  public static class IAEExceptionConversionTest
+  {
+    @Rule
+    public ExpectedException expectedException = ExpectedException.none();
+
+    private String errorMessage = "pstream connect error or disconnect/reset before header";
+    private String nullErrMsg = null;
+
+    @Test
+    public void testNullErrorMsg() throws JsonProcessingException
+    {
+      JsonParserIterator<Object> iterator = new JsonParserIterator<>(
+          JAVA_TYPE,
+          Futures.immediateFuture(
+              mockErrorResponse(nullErrMsg)
+          ),
+          URL,
+          null,
+          HOST,
+          OBJECT_MAPPER
+      );
+
+      expectedException.expect(QueryInterruptedException.class);
+      expectedException.expectMessage("");
+      iterator.hasNext();
+    }
+
+    @Test
+    public void testParsingError() throws JsonProcessingException
+    {
+      JsonParserIterator<Object> iterator = new JsonParserIterator<>(
+          JAVA_TYPE,
+          Futures.immediateFuture(
+              mockErrorResponse(errorMessage)
+          ),
+          URL,
+          null,
+          HOST,
+          OBJECT_MAPPER
+      );
+
+      expectedException.expect(QueryInterruptedException.class);
+      expectedException.expectMessage(errorMessage);
+      iterator.hasNext();
+    }
+  }
+
   private static InputStream mockErrorResponse(Exception e) throws JsonProcessingException
   {
     return new ByteArrayInputStream(OBJECT_MAPPER.writeValueAsBytes(e));
   }
+
+  private static InputStream mockErrorResponse(String errMsg) throws JsonProcessingException
+  {
+    return new ByteArrayInputStream(OBJECT_MAPPER.writeValueAsBytes(errMsg));
+  }
 }