Merge pull request #1907 from ClickHouse/v2_fix_multiline_error_msg

[client-v2] Fixed reading server errors messages if they are multi-line

Author: chernser

clickhouse-client/src/test/java/com/clickhouse/client/ClickHouseServerForTest.java

@@ -172,6 +172,7 @@ public class ClickHouseServerForTest {
                             ClickHouseProtocol.TCP.getDefaultSecurePort(),
                             ClickHouseProtocol.POSTGRESQL.getDefaultPort())
                     .withClasspathResourceMapping("containers/clickhouse-server", customDirectory, BindMode.READ_ONLY)
+                    .withClasspathResourceMapping("empty.csv", "/var/lib/clickhouse/user_files/empty.csv", BindMode.READ_ONLY)
                     .withFileSystemBind(System.getProperty("java.io.tmpdir"), getClickHouseContainerTmpDir(),
                             BindMode.READ_WRITE)
                     .withNetwork(network)

clickhouse-client/src/test/resources/empty.csv

@@ -66,6 +66,7 @@
             <artifactId>jackson-core</artifactId>
             <version>2.17.2</version>
         </dependency>
+
         <!-- Test Dependencies -->
         <dependency>
             <groupId>com.google.code.gson</groupId>

client-v2/pom.xml

@@ -63,7 +63,6 @@
 import java.net.NoRouteToHostException;
 import java.net.URI;
 import java.net.URISyntaxException;
-import java.net.URLEncoder;
 import java.net.UnknownHostException;
 import java.nio.charset.StandardCharsets;
 import java.security.NoSuchAlgorithmException;
@@ -288,36 +287,49 @@ public Exception readError(ClassicHttpResponse httpResponse) {
 
             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);
                 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));
+                for (int i = 0; i < rBytes; i++) {
+                    if (buffer[i] == lookUpStr[0]) {
+                        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) {
+                            msgBuilder.append(new String(buffer, i, rBytes - i, StandardCharsets.UTF_8));
+                            break;
+                        }
                     }
                 }
+
+                if (found) {
+                    break;
+                }
             }
 
+            while (true) {
+                int rBytes = body.read(buffer);
+                if (rBytes == -1) {
+                    break;
+                }
+                msgBuilder.append(new String(buffer, 0, rBytes, StandardCharsets.UTF_8));
+            }
+
+            String msg = msgBuilder.toString().replaceAll("\\s+", " ").replaceAll("\\\\n", " ")
+                    .replaceAll("\\\\/", "/");
+            return new ServerException(serverCode, msg);
+        } 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>");
-        } catch (IOException e) {
-            throw new ClientException("Failed to read response body", e);
         }
     }
 
@@ -504,26 +516,28 @@ private void addQueryParams(URIBuilder req, Map<String, String> chConfig, Map<St
     }
 
     private HttpEntity wrapEntity(HttpEntity httpEntity, int httpStatus, boolean isResponse) {
-
-        switch (httpStatus) {
-            case HttpStatus.SC_OK:
-            case HttpStatus.SC_CREATED:
-            case HttpStatus.SC_ACCEPTED:
-            case HttpStatus.SC_NO_CONTENT:
-            case HttpStatus.SC_PARTIAL_CONTENT:
-            case HttpStatus.SC_RESET_CONTENT:
-            case HttpStatus.SC_NOT_MODIFIED:
-            case HttpStatus.SC_BAD_REQUEST:
-                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) {
+        boolean serverCompression = MapUtils.getFlag(chConfiguration, ClickHouseClientOption.COMPRESS.getKey(), false);
+        boolean clientCompression = MapUtils.getFlag(chConfiguration, ClickHouseClientOption.DECOMPRESS.getKey(),  false);
+
+        if (serverCompression || clientCompression) {
+            // Server doesn't compress certain errors like 403
+            switch (httpStatus) {
+                case HttpStatus.SC_OK:
+                case HttpStatus.SC_CREATED:
+                case HttpStatus.SC_ACCEPTED:
+                case HttpStatus.SC_NO_CONTENT:
+                case HttpStatus.SC_PARTIAL_CONTENT:
+                case HttpStatus.SC_RESET_CONTENT:
+                case HttpStatus.SC_NOT_MODIFIED:
+                case HttpStatus.SC_BAD_REQUEST:
+                case HttpStatus.SC_INTERNAL_SERVER_ERROR:
+                    boolean useHttpCompression = MapUtils.getFlag(chConfiguration, "client.use_http_compression", false);
                     return new LZ4Entity(httpEntity, useHttpCompression, serverCompression, clientCompression,
                             MapUtils.getInt(chConfiguration, "compression.lz4.uncompressed_buffer_size"), isResponse);
-                }
-            default:
-                return httpEntity;
+            }
         }
+
+        return httpEntity;
     }
 
     public static int getHeaderInt(Header header, int defaultValue) {

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

@@ -38,6 +38,7 @@
 import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
 import java.util.Base64;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.CompletableFuture;
@@ -308,15 +309,14 @@ public static Object[][] noResponseFailureProvider() {
     }
 
     @Test(groups = { "integration" }, dataProvider = "testServerErrorHandlingDataProvider")
-    public void testServerErrorHandling(ClickHouseFormat format) {
+    public void testServerErrorHandling(ClickHouseFormat format, boolean serverCompression, boolean useHttpCompression) {
         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"))
+                .compressServerResponse(serverCompression)
+                .useHttpCompression(useHttpCompression)
                 .build()) {
 
             QuerySettings querySettings = new QuerySettings().setFormat(format);
@@ -329,17 +329,62 @@ public void testServerErrorHandling(ClickHouseFormat format) {
                 Assert.assertTrue(e.getMessage().startsWith("Code: 62. DB::Exception: Syntax error (Multi-statements are not allowed): failed at position 15 (end of query)"),
                         "Unexpected error message: " + e.getMessage());
             }
+
+
+            try (QueryResponse response = client.query("CREATE TABLE table_from_csv AS SELECT * FROM file('empty.csv')", querySettings)
+                    .get(1, TimeUnit.SECONDS)) {
+                Assert.fail("Expected exception");
+            } catch (ServerException e) {
+                e.printStackTrace();
+                Assert.assertEquals(e.getCode(), 636);
+                Assert.assertTrue(e.getMessage().startsWith("Code: 636. DB::Exception: The table structure cannot be extracted from a CSV format file. Error: The table structure cannot be extracted from a CSV format file: the file is empty. You can specify the structure manually: (in file/uri /var/lib/clickhouse/user_files/empty.csv). (CANNOT_EXTRACT_TABLE_STRUCTURE)"),
+                        "Unexpected error message: " + e.getMessage());
+            }
+
+
         } catch (Exception e) {
             e.printStackTrace();
             Assert.fail(e.getMessage(), e);
         }
+
+        try (Client client = new Client.Builder()
+                .addEndpoint(server.getBaseUri())
+                .setUsername("non-existing-user")
+                .setPassword("nothing")
+                .compressServerResponse(serverCompression)
+                .useHttpCompression(useHttpCompression)
+                .build()) {
+
+            try (QueryResponse response = client.query("SELECT 1").get(1, TimeUnit.SECONDS)) {
+                Assert.fail("Expected exception");
+            } catch (ServerException e) {
+                e.printStackTrace();
+                Assert.assertEquals(e.getCode(), 516);
+                Assert.assertTrue(e.getMessage().startsWith("Code: 516. DB::Exception: non-existing-user: Authentication failed: password is incorrect, or there is no user with such name. (AUTHENTICATION_FAILED)"),
+                        e.getMessage());
+            } catch (Exception e) {
+                e.printStackTrace();
+                Assert.fail("Unexpected exception", e);
+            }
+        }
     }
 
     @DataProvider(name = "testServerErrorHandlingDataProvider")
-    public static Object[] testServerErrorHandlingDataProvider() {
-        return new Object[] { ClickHouseFormat.JSON, ClickHouseFormat.TabSeparated, ClickHouseFormat.RowBinary };
-    }
+    public static Object[][] testServerErrorHandlingDataProvider() {
+        EnumSet<ClickHouseFormat> formats = EnumSet.of(ClickHouseFormat.CSV, ClickHouseFormat.TSV,
+                                            ClickHouseFormat.JSON, ClickHouseFormat.JSONCompact);
 
+        Object[][] result = new Object[formats.size() * 3][];
+
+        int i = 0;
+        for (ClickHouseFormat format : formats) {
+            result[i++] = new Object[]{format, false, false};
+            result[i++] = new Object[]{format, true, false};
+            result[i++] = new Object[]{format, true, true};
+        }
+
+        return result;
+    }
 
     @Test(groups = { "integration" })
     public void testErrorWithSuccessfulResponse() {
@@ -377,6 +422,61 @@ public void testErrorWithSuccessfulResponse() {
         }
     }
 
+    @Test(groups = { "integration" }, dataProvider = "testServerErrorsUncompressedDataProvider")
+    public void testServerErrorsUncompressed(int code, String message, String expectedMessage) {
+        WireMockServer mockServer = new WireMockServer( WireMockConfiguration
+                .options().port(9090).notifier(new ConsoleNotifier(false)));
+        mockServer.start();
+
+        mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
+                .willReturn(WireMock.aResponse()
+                        .withStatus(HttpStatus.SC_OK)
+                        .withChunkedDribbleDelay(2, 200)
+                        .withHeader("X-ClickHouse-Exception-Code", String.valueOf(code))
+                        .withHeader("X-ClickHouse-Summary",
+                                "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")
+                        .withBody(message))
+                .build());
+
+        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
+                .setUsername("default")
+                .setPassword("")
+                .compressServerResponse(false)
+                .build()) {
+
+            try (QueryResponse response = client.query("SELECT 1").get(1, TimeUnit.SECONDS)) {
+                Assert.fail("Expected exception");
+            } catch (ServerException e) {
+                e.printStackTrace();
+                Assert.assertEquals(e.getCode(), code);
+                Assert.assertEquals(e.getMessage(), expectedMessage);
+            } catch (Exception e) {
+                e.printStackTrace();
+                Assert.fail("Unexpected exception", e);
+            }
+        } finally {
+            mockServer.stop();
+        }
+    }
+
+    @DataProvider(name = "testServerErrorsUncompressedDataProvider")
+    public static Object[][] testServerErrorsUncompressedDataProvider() {
+        return new Object[][] {
+                { 241, "Code: 241. DB::Exception: Memory limit (for query) exceeded: would use 97.21 MiB",
+                        "Code: 241. DB::Exception: Memory limit (for query) exceeded: would use 97.21 MiB"},
+                {900, "Code: 900. DB::Exception: \uD83D\uDCBE Floppy disk is full",
+                        "Code: 900. DB::Exception: \uD83D\uDCBE Floppy disk is full"},
+                {901, "Code: 901. DB::Exception: I write, erase, rewrite\n" +
+                        "Erase again, and then\n" +
+                        "A poppy blooms\n" +
+                        " (by Katsushika Hokusai)",
+                        "Code: 901. DB::Exception: I write, erase, rewrite " +
+                                "Erase again, and then " +
+                                "A poppy blooms" +
+                                " (by Katsushika Hokusai)"}
+        };
+    }
+
     @Test(groups = { "integration" })
     public void testAdditionalHeaders() {
         WireMockServer mockServer = new WireMockServer( WireMockConfiguration