Merge pull request #2511 from ClickHouse/fix_error_reading

[client-v2] fix reading error

Author: chernser

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

@@ -91,6 +91,14 @@ private boolean readFully(byte[] b, int off, int len) throws IOException {
         return true;
     }
 
+    public byte[] getHeaderBuffer() {
+        return headerBuff;
+    }
+
+    public InputStream getInputStream() {
+        return in;
+    }
+
     private int refill() throws IOException {
 
         // read header

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

@@ -332,7 +332,8 @@ public CloseableHttpClient createHttpClient(boolean initSslContext, Map<String,
         return clientBuilder.build();
     }
 
-    private static final String ERROR_CODE_PREFIX_PATTERN = "Code: %d. DB::Exception:";
+//    private static final String ERROR_CODE_PREFIX_PATTERN = "Code: %d. DB::Exception:";
+    private static final String ERROR_CODE_PREFIX_PATTERN = "%d. DB::Exception:";
 
     /**
      * Reads status line and if error tries to parse response body to get server error message.
@@ -342,14 +343,27 @@ public CloseableHttpClient createHttpClient(boolean initSslContext, Map<String,
      */
     public Exception readError(ClassicHttpResponse httpResponse) {
         int serverCode = getHeaderInt(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE), 0);
-        try (InputStream body = httpResponse.getEntity().getContent()) {
-
+        InputStream body = null;
+        try {
+            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);
             StringBuilder msgBuilder = new StringBuilder();
             boolean found = false;
             while (true) {
-                int rBytes = body.read(buffer);
+                int rBytes = -1;
+                try {
+                    rBytes = body.read(buffer);
+                } catch (ClientException e) {
+                    // Invalid LZ4 Magic
+                    if (body instanceof ClickHouseLZ4InputStream) {
+                        ClickHouseLZ4InputStream stream = (ClickHouseLZ4InputStream) body;
+                        body = stream.getInputStream();
+                        byte[] headerBuffer = stream.getHeaderBuffer();
+                        System.arraycopy(headerBuffer, 0, buffer, 0, headerBuffer.length);
+                        rBytes = headerBuffer.length;
+                    }
+                }
                 if (rBytes == -1) {
                     break;
                 }
@@ -388,7 +402,7 @@ public Exception readError(ClassicHttpResponse httpResponse) {
             if (msg.trim().isEmpty()) {
                 msg = String.format(ERROR_CODE_PREFIX_PATTERN, serverCode) + " <Unreadable error message> (transport error: " + httpResponse.getCode() + ")";
             }
-            return new ServerException(serverCode, msg, httpResponse.getCode());
+            return new ServerException(serverCode, "Code: " + msg, httpResponse.getCode());
         } catch (Exception e) {
             LOG.error("Failed to read error message", e);
             return new ServerException(serverCode, String.format(ERROR_CODE_PREFIX_PATTERN, serverCode) + " <Unreadable error message> (transport error: " + httpResponse.getCode() + ")", httpResponse.getCode());