fixed parsing error in response

Author: chernser

client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java

@@ -53,15 +53,20 @@
 
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLException;
+import java.io.BufferedReader;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.io.OutputStream;
 import java.net.ConnectException;
 import java.net.InetSocketAddress;
 import java.net.NoRouteToHostException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.security.NoSuchAlgorithmException;
 import java.util.Base64;
 import java.util.EnumSet;
@@ -272,18 +277,47 @@ public CloseableHttpClient createHttpClient() {
         return clientBuilder.build();
     }
 
+    private static final String ERROR_CODE_PREFIX_PATTERN = "Code: %d. DB::Exception:";
     /**
      * Reads status line and if error tries to parse response body to get server error message.
      *
      * @param httpResponse - HTTP response
      * @return
      */
     public Exception readError(ClassicHttpResponse httpResponse) {
-        try (ByteArrayOutputStream out = new ByteArrayOutputStream(ERROR_BODY_BUFFER_SIZE)) {
-            httpResponse.getEntity().writeTo(out);
-            String message = out.toString();
-            int serverCode = getHeaderInt(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE), 0);
-            return new ServerException(serverCode, message);
+        int serverCode = getHeaderInt(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE), 0);
+        try (InputStream body = httpResponse.getEntity().getContent()) {
+
+            byte [] buffer = new byte[ERROR_BODY_BUFFER_SIZE];
+            byte [] lookUpStr = String.format(ERROR_CODE_PREFIX_PATTERN, serverCode).getBytes(StandardCharsets.UTF_8);
+            while (true) {
+                int rBytes = body.read(buffer);
+                if (rBytes == -1) {
+                    break;
+                } else {
+                    for (int i = 0; i < rBytes; i++) {
+                        if (buffer[i] == lookUpStr[0]) {
+                            boolean found = true;
+                            for (int j = 1; j < Math.min(rBytes - i, lookUpStr.length); j++) {
+                                if (buffer[i + j] != lookUpStr[j]) {
+                                    found = false;
+                                    break;
+                                }
+                            }
+                            if (found) {
+                                int start = i;
+                                while (i < rBytes && buffer[i] != '\n') {
+                                    i++;
+                                }
+
+                                return new ServerException(serverCode, new String(buffer, start, i -start, StandardCharsets.UTF_8));
+                            }
+                        }
+                    }
+                }
+            }
+
+            return new ServerException(serverCode, ERROR_CODE_PREFIX_PATTERN.formatted(serverCode) + " <Unreadable error message>");
         } catch (IOException e) {
             throw new ClientException("Failed to read response body", e);
         }
@@ -307,12 +341,13 @@ public ClassicHttpResponse executeRequest(ClickHouseNode server, Map<String, Obj
                 .build();
         req.setConfig(httpReqConfig);
         // setting entity. wrapping if compression is enabled
-        req.setEntity(wrapEntity(new EntityTemplate(-1, CONTENT_TYPE, null, writeCallback)));
+        req.setEntity(wrapEntity(new EntityTemplate(-1, CONTENT_TYPE, null, writeCallback), false));
 
         HttpClientContext context = HttpClientContext.create();
 
         try {
             ClassicHttpResponse httpResponse = httpClient.executeOpen(null, req, context);
+            httpResponse.setEntity(wrapEntity(httpResponse.getEntity(), true));
             if (httpResponse.getCode() == HttpStatus.SC_PROXY_AUTHENTICATION_REQUIRED) {
                 throw new ClientMisconfigurationException("Proxy authentication required. Please check your proxy settings.");
             } else if (httpResponse.getCode() >= HttpStatus.SC_BAD_REQUEST &&
@@ -329,8 +364,6 @@ public ClassicHttpResponse executeRequest(ClickHouseNode server, Map<String, Obj
                 httpResponse.close();
                 return httpResponse;
             }
-
-            httpResponse.setEntity(wrapEntity(httpResponse.getEntity()));
             return httpResponse;
 
         } catch (UnknownHostException e) {
@@ -413,13 +446,13 @@ private void addQueryParams(URIBuilder req, Map<String, String> chConfig, Map<St
         }
     }
 
-    private HttpEntity wrapEntity(HttpEntity httpEntity) {
+    private HttpEntity wrapEntity(HttpEntity httpEntity, boolean isResponse) {
         boolean serverCompression = chConfiguration.getOrDefault(ClickHouseClientOption.COMPRESS.getKey(), "false").equalsIgnoreCase("true");
         boolean clientCompression = chConfiguration.getOrDefault(ClickHouseClientOption.DECOMPRESS.getKey(), "false").equalsIgnoreCase("true");
         boolean useHttpCompression = chConfiguration.getOrDefault("client.use_http_compression", "false").equalsIgnoreCase("true");
         if (serverCompression || clientCompression) {
             return new LZ4Entity(httpEntity, useHttpCompression, serverCompression, clientCompression,
-                    MapUtils.getInt(chConfiguration, "compression.lz4.uncompressed_buffer_size"));
+                    MapUtils.getInt(chConfiguration, "compression.lz4.uncompressed_buffer_size"), isResponse);
         } else {
             return httpEntity;
         }

client-v2/src/main/java/com/clickhouse/client/api/internal/LZ4Entity.java

@@ -16,55 +16,70 @@
 class LZ4Entity implements HttpEntity {
 
     private HttpEntity httpEntity;
-    
+
     private final boolean useHttpCompression;
-    private final boolean serverCompression;
-    private final boolean clientCompression;
-    
+
     private final int bufferSize;
 
+    private final boolean isResponse;
+
+    private boolean serverCompression;
+
+    private boolean clientCompression;
+
     LZ4Entity(HttpEntity httpEntity, boolean useHttpCompression, boolean serverCompression, boolean clientCompression,
-              int bufferSize) {
+              int bufferSize, boolean isResponse) {
         this.httpEntity = httpEntity;
         this.useHttpCompression = useHttpCompression;
+        this.bufferSize = bufferSize;
         this.serverCompression = serverCompression;
         this.clientCompression = clientCompression;
-        this.bufferSize = bufferSize;
+        this.isResponse = isResponse;
     }
-    
+
     @Override
     public boolean isRepeatable() {
         return httpEntity.isRepeatable();
     }
 
     @Override
     public InputStream getContent() throws IOException, UnsupportedOperationException {
-        if (serverCompression && useHttpCompression) {
-            InputStream content = httpEntity.getContent();
-            try {
-                return new FramedLZ4CompressorInputStream(content);
-            } catch (IOException e) {
-                // This is the easiest way to handle empty content because
-                // - streams at this point wrapped with something else and we can't check content length
-                // - exception is thrown with no details
-                // So we just return original content and if there is a real data in it we will get error later
-                return content;
-            }
+        if (!isResponse && serverCompression) {
+            throw new UnsupportedOperationException("Unsupported: getting compressed content of request");
         } else if (serverCompression) {
-            return new ClickHouseLZ4InputStream(httpEntity.getContent(), LZ4Factory.fastestInstance().fastDecompressor(),
-                    bufferSize);
+            if (useHttpCompression) {
+                InputStream content = httpEntity.getContent();
+                try {
+                    return new FramedLZ4CompressorInputStream(content);
+                } catch (IOException e) {
+                    // This is the easiest way to handle empty content because
+                    // - streams at this point wrapped with something else and we can't check content length
+                    // - exception is thrown with no details
+                    // So we just return original content and if there is a real data in it we will get error later
+                    return content;
+                }
+            } else  {
+                return new ClickHouseLZ4InputStream(httpEntity.getContent(), LZ4Factory.fastestInstance().fastDecompressor(),
+                        bufferSize);
+            }
         } else {
             return httpEntity.getContent();
         }
     }
 
     @Override
     public void writeTo(OutputStream outStream) throws IOException {
-        if (clientCompression && useHttpCompression) {
-            httpEntity.writeTo(new FramedLZ4CompressorOutputStream(outStream));
+        if (isResponse && serverCompression) {
+            // called by us to get compressed response
+            throw new UnsupportedOperationException("Unsupported: writing compressed response to elsewhere");
         } else if (clientCompression) {
-            httpEntity.writeTo(new ClickHouseLZ4OutputStream(outStream, LZ4Factory.fastestInstance().fastCompressor(),
-                    bufferSize));
+            // called by client to send data
+            if (useHttpCompression) {
+                httpEntity.writeTo(new FramedLZ4CompressorOutputStream(outStream));
+            } else {
+                httpEntity.writeTo(new ClickHouseLZ4OutputStream(outStream, LZ4Factory.fastestInstance().fastCompressor(),
+                        bufferSize));
+            }
         } else {
             httpEntity.writeTo(outStream);
         }

client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java

@@ -5,11 +5,14 @@
 import com.clickhouse.client.api.ClientFaultCause;
 import com.clickhouse.client.api.ConnectionInitiationException;
 import com.clickhouse.client.api.ConnectionReuseStrategy;
+import com.clickhouse.client.api.ServerException;
+import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader;
 import com.clickhouse.client.api.enums.Protocol;
 import com.clickhouse.client.api.enums.ProxyType;
 import com.clickhouse.client.api.insert.InsertResponse;
 import com.clickhouse.client.api.query.GenericRecord;
 import com.clickhouse.client.api.query.QueryResponse;
+import com.clickhouse.client.api.query.QuerySettings;
 import com.clickhouse.client.config.ClickHouseClientOption;
 import com.clickhouse.data.ClickHouseFormat;
 import com.github.tomakehurst.wiremock.WireMockServer;
@@ -304,4 +307,38 @@ public static Object[][] noResponseFailureProvider() {
                 {selectBody, 0, queryFunction, true}
         };
     }
+
+    @Test(groups = { "integration" }, dataProvider = "testServerErrorHandlingDataProvider")
+    public void testServerErrorHandling(ClickHouseFormat format) {
+        ClickHouseNode server = getServer(ClickHouseProtocol.HTTP);
+        try (Client client = new Client.Builder()
+                .addEndpoint(server.getBaseUri())
+                .setUsername("default")
+                .setPassword("")
+                .useNewImplementation(true)
+                // TODO: fix in old client
+//                .useNewImplementation(System.getProperty("client.tests.useNewImplementation", "false").equals("true"))
+                .build()) {
+
+            QuerySettings querySettings = new QuerySettings().setFormat(format);
+            try (QueryResponse response =
+                         client.query("SELECT invalid;statement", querySettings).get(1, TimeUnit.SECONDS)) {
+                Assert.fail("Expected exception");
+            } catch (ClientException e) {
+                e.printStackTrace();
+                ServerException serverException = (ServerException) e.getCause();
+                Assert.assertEquals(serverException.getCode(), 62);
+                Assert.assertTrue(serverException.getMessage().startsWith("Code: 62. DB::Exception: Syntax error (Multi-statements are not allowed): failed at position 15 (end of query)"),
+                        "Unexpected error message: " + serverException.getMessage());
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            Assert.fail(e.getMessage(), e);
+        }
+    }
+
+    @DataProvider(name = "testServerErrorHandlingDataProvider")
+    public static Object[] testServerErrorHandlingDataProvider() {
+        return new Object[] { ClickHouseFormat.JSON, ClickHouseFormat.TabSeparated, ClickHouseFormat.RowBinary };
+    }
 }