HttpTransportTests.java

package com.clickhouse.client;

import com.clickhouse.client.api.Client;
import com.clickhouse.client.api.ClientConfigProperties;
import com.clickhouse.client.api.ClientException;
import com.clickhouse.client.api.Session;
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.command.CommandResponse;
import com.clickhouse.client.api.command.CommandSettings;
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.InsertSettings;
import com.clickhouse.client.api.insert.InsertResponse;
import com.clickhouse.client.api.internal.DataTypeConverter;
import com.clickhouse.client.api.internal.ServerSettings;
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;
import com.github.tomakehurst.wiremock.client.WireMock;
import com.github.tomakehurst.wiremock.common.ConsoleNotifier;
import com.github.tomakehurst.wiremock.common.Slf4jNotifier;
import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
import com.github.tomakehurst.wiremock.http.Fault;
import com.github.tomakehurst.wiremock.http.trafficlistener.WiremockNetworkTrafficListener;
import org.apache.hc.core5.http.ConnectionClosedException;
import org.apache.hc.core5.http.ConnectionRequestTimeoutException;
import org.apache.hc.core5.http.HttpHeaders;
import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.net.URIBuilder;
import org.testcontainers.utility.ThrowingFunction;
import org.testng.Assert;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.net.InetAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.temporal.ChronoUnit;
import java.util.Arrays;
import java.util.Base64;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;
import java.util.zip.GZIPOutputStream;

import static com.github.tomakehurst.wiremock.stubbing.Scenario.STARTED;
import static org.testng.Assert.fail;

@Test(groups = {"integration"})
public class HttpTransportTests extends BaseIntegrationTest {

    @Test(groups = {"integration"},dataProvider = "testConnectionTTLProvider")
    @SuppressWarnings("java:S2925")
    public void testConnectionTTL(Long connectionTtl, Long keepAlive, int openSockets) throws Exception {
        if (isCloud()) {
            return; // skip cloud tests because of wiremock proxy. TODO: fix it
        }
        ClickHouseNode server = getServer(ClickHouseProtocol.HTTP);

        int proxyPort = new Random().nextInt(1000) + 10000;
        ConnectionCounterListener connectionCounter = new ConnectionCounterListener();
        WireMockServer proxy = new WireMockServer(WireMockConfiguration
                .options().port(proxyPort)
                .networkTrafficListener(connectionCounter)
                .notifier(new Slf4jNotifier(true)));
        proxy.start();
        URIBuilder targetURI = new URIBuilder(server.getBaseUri())
                .setPath("");
        proxy.addStubMapping(WireMock.post(WireMock.anyUrl())
                .willReturn(WireMock.aResponse().proxiedFrom(targetURI.build().toString())).build());

        Client.Builder clientBuilder = new Client.Builder()
                .addEndpoint(server.getBaseUri())
                .setUsername("default")
                .setPassword(getPassword())
                .addProxy(ProxyType.HTTP, "localhost", proxyPort);
        if (connectionTtl != null) {
            clientBuilder.setConnectionTTL(connectionTtl, ChronoUnit.MILLIS);
        }
        if (keepAlive != null) {
            clientBuilder.setKeepAliveTimeout(keepAlive, ChronoUnit.MILLIS);
        }

        try (Client client = clientBuilder.build()) {
            List<GenericRecord> resp = client.queryAll("select 1");
            Assert.assertEquals(resp.stream().findFirst().get().getString(1), "1");

            try {
                Thread.sleep(1000L);
            } catch (InterruptedException e) {
                Assert.fail("Unexpected exception", e);
            }

            resp = client.queryAll("select 1");
            Assert.assertEquals(resp.stream().findFirst().get().getString(1), "1");
        } catch (Exception e) {
            e.printStackTrace();
            Assert.fail("Unexpected exception", e);
        } finally {
            Assert.assertEquals(connectionCounter.opened.get(), openSockets);
            proxy.stop();
        }
    }

    @DataProvider(name = "testConnectionTTLProvider")
    public static Object[][]  testConnectionTTLProvider() {
        return new Object[][] {
                { 1000L, null, 2 },
                { 2000L, null, 1 },
                { null, 2000L, 1 },
                { null, 500L, 2 },
                { 1000L, 0L, 2 },
                { 1000L, 3000L, 2}
        };
    }

    private static class ConnectionCounterListener implements WiremockNetworkTrafficListener {

        private AtomicInteger opened = new AtomicInteger(0);
        private AtomicInteger closed = new AtomicInteger(0);

        @Override
        public void opened(Socket socket) {
            opened.incrementAndGet();
        }

        @Override
        public void incoming(Socket socket, ByteBuffer bytes) {
            // ignore
        }

        @Override
        public void outgoing(Socket socket, ByteBuffer bytes) {
            // ignore
        }

        @Override
        public void closed(Socket socket) {
            closed.incrementAndGet();
        }
    }

    @Test(groups = {"integration"})
    public void testConnectionRequestTimeout() {
        if (isCloud()) {
            return; // mocked server
        }

        int serverPort = new Random().nextInt(1000) + 10000;
        ConnectionCounterListener connectionCounter = new ConnectionCounterListener();
        WireMockServer proxy = new WireMockServer(WireMockConfiguration
                .options().port(serverPort)
                .networkTrafficListener(connectionCounter)
                .notifier(new Slf4jNotifier(true)));
        proxy.start();
        proxy.addStubMapping(WireMock.post(WireMock.anyUrl())
                .willReturn(WireMock.aResponse().withFixedDelay(5000)
                        .withStatus(HttpStatus.SC_NOT_FOUND)).build());

        Client.Builder clientBuilder = new Client.Builder()
                .addEndpoint("http://localhost:" + serverPort)
                .setUsername("default")
                .setPassword(getPassword())
                .retryOnFailures(ClientFaultCause.None)
                .setMaxConnections(1)
                .setOption(ClickHouseClientOption.ASYNC.getKey(), "true")
                .setSocketTimeout(10000, ChronoUnit.MILLIS)
                .setConnectionRequestTimeout(5, ChronoUnit.MILLIS);

        try (Client client = clientBuilder.build()) {
            CompletableFuture<QueryResponse> f1 = client.query("select 1");
            Thread.sleep(500L);
            CompletableFuture<QueryResponse> f2 = client.query("select 1");
            f2.get();
        } catch (ExecutionException e) {
            e.printStackTrace();
            Assert.assertEquals(e.getCause().getClass(), ConnectionInitiationException.class);
            Assert.assertEquals(e.getCause().getCause().getClass(), ConnectionRequestTimeoutException.class);
        } catch (Exception e) {
            e.printStackTrace();
            Assert.fail("Unexpected exception", e);
        } finally {
            proxy.stop();
        }
    }

    @Test(groups = {"integration"})
    public void testConnectionReuseStrategy() {
        if (isCloud()) {
            return; // mocked server
        }

        ClickHouseNode server = getServer(ClickHouseProtocol.HTTP);

        try (Client client = new Client.Builder()
                .addEndpoint(server.getBaseUri())
                .setUsername("default")
                .setPassword(getPassword())
                .setConnectionReuseStrategy(ConnectionReuseStrategy.LIFO)
                .build()) {

            List<GenericRecord> records = client.queryAll("SELECT timezone()");
            Assert.assertTrue(records.size() > 0);
            Assert.assertEquals(records.get(0).getString(1), "UTC");
        } catch (Exception e) {
            e.printStackTrace();
            Assert.fail(e.getMessage());
        }
    }

    @Test(groups = { "integration" })
    public void testSecureConnection() {
        if (isCloud()) {
            return; // will fail in other tests
        }

        ClickHouseNode secureServer = getSecureServer(ClickHouseProtocol.HTTP);

        try (Client client = new Client.Builder()
                .addEndpoint("https://localhost:" + secureServer.getPort())
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .setRootCertificate("containers/clickhouse-server/certs/localhost.crt")
                .compressClientRequest(true)
                .build()) {

            List<GenericRecord> records = client.queryAll("SELECT timezone()");
            Assert.assertTrue(records.size() > 0);
            Assert.assertEquals(records.get(0).getString(1), "UTC");
        } catch (Exception e) {
            e.printStackTrace();
            Assert.fail(e.getMessage());
        }
    }

    @Test(groups = { "integration" }, dataProvider = "NoResponseFailureProvider")
    public void testInsertAndNoHttpResponseFailure(String body, int maxRetries, ThrowingFunction<Client, Void> function,
                                                   boolean shouldFail) {
        if (isCloud()) {
            return; // mocked server
        }

        WireMockServer faultyServer = new WireMockServer( WireMockConfiguration
                .options().port(9090).notifier(new ConsoleNotifier(false)));
        faultyServer.start();

        // First request gets no response
        faultyServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                .withRequestBody(WireMock.equalTo(body))
                .inScenario("Retry")
                .whenScenarioStateIs(STARTED)
                .willSetStateTo("Failed")
                .willReturn(WireMock.aResponse().withFault(Fault.EMPTY_RESPONSE)).build());

        // Second request gets a response (retry)
        faultyServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                .withRequestBody(WireMock.equalTo(body))
                .inScenario("Retry")
                .whenScenarioStateIs("Failed")
                .willSetStateTo("Done")
                .willReturn(WireMock.aResponse()
                        .withHeader("X-ClickHouse-Summary",
                                "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());

        Client mockServerClient = new Client.Builder()
                .addEndpoint(Protocol.HTTP, "localhost", faultyServer.port(), false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .compressClientRequest(false)
                .setMaxRetries(maxRetries)
                .build();

        try {
            function.apply(mockServerClient);
        } catch (ConnectionInitiationException e) {
            e.printStackTrace();
            if (!shouldFail) {
                Assert.fail("Unexpected exception", e);
            }
            return;
        } catch (Exception e) {
            Assert.fail("Unexpected exception", e);
        } finally {
            faultyServer.stop();
        }

        if (shouldFail) {
            Assert.fail("Expected exception");
        }
    }

    @DataProvider(name = "NoResponseFailureProvider")
    public static Object[][] noResponseFailureProvider() {

        String insertBody = "1\t2\t3\n";
        ThrowingFunction<Client, Void> insertFunction = (client) -> {
            InsertResponse insertResponse = client.insert("table01",
                    new ByteArrayInputStream("1\t2\t3\n".getBytes()), ClickHouseFormat.TSV).get(30, TimeUnit.SECONDS);
            insertResponse.close();
            return null;
        };

        String selectBody = "select timezone()";
        ThrowingFunction<Client, Void> queryFunction = (client) -> {
            QueryResponse response = client.query("select timezone()").get(30, TimeUnit.SECONDS);
            response.close();
            return null;
        };

        return new Object[][]{
                {insertBody, 1, insertFunction, false},
                {selectBody, 1, queryFunction, false},
                {insertBody, 0, insertFunction, true},
                {selectBody, 0, queryFunction, true}
        };
    }

    @Test(groups = { "integration" }, dataProvider = "testServerErrorHandlingDataProvider")
    public void testServerErrorHandling(ClickHouseFormat format, boolean serverCompression, boolean useHttpCompression) {
        if (isCloud()) {
            return; // mocked server
        }

        ClickHouseNode server = getServer(ClickHouseProtocol.HTTP);
        try (Client client = new Client.Builder()
                .addEndpoint(server.getBaseUri())
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .compressServerResponse(serverCompression)
                .useHttpCompression(useHttpCompression)
                .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 (ServerException e) {
                e.printStackTrace();
                Assert.assertEquals(e.getCode(), 62);
                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 ENGINE MergeTree ORDER BY () 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().contains("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());
            }

            querySettings.serverSetting("unknown_setting", "1");
            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(), 115);
                Assert.assertTrue(e.getMessage().startsWith("Code: 115. DB::Exception: Setting unknown_setting is neither a builtin setting nor started with the prefix 'custom_' registered for user-defined settings. (UNKNOWN_SETTING)"),
                        "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);
            }
        }

        if (!isCloud()) {
            Client.Builder clientBuilder = new Client.Builder()
                    .addEndpoint("http://" + server.getHost() + ":" + server.getPort() + "/some-path")
                    .setUsername("default")
                    .setPassword(ClickHouseServerForTest.getPassword())
                    .compressServerResponse(serverCompression)
                    .useHttpCompression(useHttpCompression)
                    .setDefaultDatabase(ClickHouseServerForTest.getDatabase())
                    .serverSetting(ServerSettings.WAIT_END_OF_QUERY, "1");

            try (Client client = clientBuilder.build()) {
                client.queryAll("select 1");
                fail("Exception expected");
            } catch (ClientException e) {
                Assert.assertTrue(e.getCause().getMessage().contains("no handle /some-path?"));
            }
        }
    }

    @DataProvider(name = "testServerErrorHandlingDataProvider")
    public static Object[][] testServerErrorHandlingDataProvider() {
        EnumSet<ClickHouseFormat> formats = EnumSet.of(ClickHouseFormat.CSV, ClickHouseFormat.TSV,
                                            ClickHouseFormat.JSON, ClickHouseFormat.JSONCompact);

        int permutations = 3;
        Object[][] result = new Object[formats.size() * permutations][];

        int i = 0;
        for (ClickHouseFormat format : formats) {
            // format, server compression, http compression
            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() {
        WireMockServer mockServer = new WireMockServer( WireMockConfiguration
                .options().port(9090).notifier(new ConsoleNotifier(false)));
        mockServer.start();

        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .compressServerResponse(false)
                .build()) {
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .willReturn(WireMock.aResponse()
                            .withStatus(HttpStatus.SC_OK)
                            .withChunkedDribbleDelay(2, 200)
                            .withHeader("X-ClickHouse-Exception-Code", "241")
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")
                            .withBody("Code: 241. DB::Exception: Memory limit (for query) exceeded: would use 97.21 MiB"))
                    .build());

            try (QueryResponse response = client.query("SELECT 1").get(1, TimeUnit.SECONDS)) {
                Assert.fail("Expected exception");
            } catch (ServerException e) {
                e.printStackTrace();
                Assert.assertTrue(e.getMessage().startsWith("Code: 241. DB::Exception: Memory limit (for query) exceeded: would use 97.21 MiB"));
            } catch (Exception e) {
                e.printStackTrace();
                Assert.fail("Unexpected exception", e);
            }
        } finally {
            mockServer.stop();
        }
    }

    @Test(groups = { "integration" }, dataProvider = "testServerErrorsUncompressedDataProvider")
    public void testServerErrorsUncompressed(int code, String message, String expectedMessage) {
        if (isCloud()) {
            return; // mocked server
        }

        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(ClickHouseServerForTest.getPassword())
                .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.assertTrue(e.getMessage().startsWith(expectedMessage), "but started with " + e.getMessage());
            } 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() {
        if (isCloud()) {
            return; // mocked server
        }

        WireMockServer mockServer = new WireMockServer( WireMockConfiguration
                .options().port(9090).notifier(new ConsoleNotifier(false)));
        mockServer.start();


        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .httpHeader("X-ClickHouse-Test", "default_value")
                .httpHeader("X-ClickHouse-Test-2", Arrays.asList("default_value1", "default_value2"))
                .httpHeader("X-ClickHouse-Test-3", Arrays.asList("default_value1", "default_value2"))
                .httpHeader("X-ClickHouse-Test-4", "default_value4")
                .build()) {
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .withHeader("X-ClickHouse-Test", WireMock.equalTo("test"))
                    .withHeader("X-ClickHouse-Test-2", WireMock.equalTo("test1,test2"))
                    .withHeader("X-ClickHouse-Test-3", WireMock.equalTo("default_value1,default_value2"))
                    .withHeader("X-ClickHouse-Test-4", WireMock.equalTo("default_value4"))

                    .willReturn(WireMock.aResponse()
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());

            QuerySettings querySettings = new QuerySettings()
                    .httpHeader("X-ClickHouse-Test", "test")
                    .httpHeader("X-ClickHouse-Test-2", Arrays.asList("test1", "test2"));

            try (QueryResponse response = client.query("SELECT 1", querySettings).get(10, TimeUnit.SECONDS)) {
                Assert.assertEquals(response.getReadBytes(), 10);
            } catch (Exception e) {
                e.printStackTrace();
                Assert.fail("Unexpected exception", e);
            }
        } finally {
            mockServer.stop();
        }
    }

    @Test(groups = { "integration" })
    public void testServerSettings() {
        if (isCloud()) {
            return; // mocked server
        }

        WireMockServer mockServer = new WireMockServer( WireMockConfiguration
                .options().port(9090).notifier(new ConsoleNotifier(false)));
        mockServer.start();

        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .serverSetting("max_threads", "10")
                .serverSetting("async_insert", "1")
                .serverSetting("roles", Arrays.asList("role1", "role2"))
                .compressClientRequest(true)
                .build()) {

            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                            .withQueryParam("max_threads", WireMock.equalTo("10"))
                            .withQueryParam("async_insert", WireMock.equalTo("3"))
                            .withQueryParam("roles", WireMock.equalTo("role3,role2"))
                            .withQueryParam("compress", WireMock.equalTo("0"))
                    .willReturn(WireMock.aResponse()
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());

            QuerySettings querySettings = new QuerySettings()
                    .serverSetting("max_threads", "10")
                    .serverSetting("async_insert", "3")
                    .serverSetting("roles", Arrays.asList("role3", "role2"))
                    .serverSetting("compress", "0");
            try (QueryResponse response = client.query("SELECT 1", querySettings).get(1, TimeUnit.SECONDS)) {
                Assert.assertEquals(response.getReadBytes(), 10);
            } catch (Exception e) {
                e.printStackTrace();
                Assert.fail("Unexpected exception", e);
            } finally {
                mockServer.stop();
            }
        }
    }

    @Test(groups = { "integration" })
    public void testSessionSettingsClientAndOperationLevels() {
        if (isCloud()) {
            return; // mocked server
        }

        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().dynamicPort().notifier(new ConsoleNotifier(false)));
        mockServer.start();

        Session clientSession = new Session()
                .setSessionId("client-session")
                .setSessionCheck(false)
                .setSessionTimeout(60)
                .setSessionTimezone("Europe/London");

        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .use(clientSession)
                .compressClientRequest(false)
                .build()) {
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .withQueryParam("session_id", WireMock.equalTo("query-session"))
                    .withQueryParam("session_check", WireMock.equalTo("1"))
                    .withQueryParam("session_timeout", WireMock.equalTo("15"))
                    .withQueryParam("session_timezone", WireMock.equalTo("Asia/Tokyo"))
                    .willReturn(WireMock.aResponse()
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());

            Session querySession = new Session()
                    .setSessionId("query-session")
                    .setSessionCheck(true)
                    .setSessionTimeout(15)
                    .setSessionTimezone("Asia/Tokyo");
            QuerySettings querySettings = new QuerySettings().use(querySession);
            try (QueryResponse response = client.query("SELECT 1", querySettings).get(1, TimeUnit.SECONDS)) {
                Assert.assertEquals(response.getReadBytes(), 10);
            }

            mockServer.resetRequests();
            mockServer.resetMappings();
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .withQueryParam("session_id", WireMock.equalTo("client-session"))
                    .withQueryParam("session_check", WireMock.equalTo("0"))
                    .withQueryParam("session_timeout", WireMock.equalTo("60"))
                    .withQueryParam("session_timezone", WireMock.equalTo("Europe/London"))
                    .willReturn(WireMock.aResponse()
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"11\", \"read_rows\": \"1\"}")).build());

            try (CommandResponse response = client.execute("SELECT 1").get(1, TimeUnit.SECONDS)) {
                Assert.assertEquals(response.getReadBytes(), 11);
            }

            mockServer.resetRequests();
            mockServer.resetMappings();
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .withQueryParam("session_id", WireMock.equalTo("insert-session"))
                    .withQueryParam("session_check", WireMock.equalTo("0"))
                    .withQueryParam("session_timeout", WireMock.equalTo("90"))
                    .withQueryParam("session_timezone", WireMock.equalTo("America/Denver"))
                    .willReturn(WireMock.aResponse()
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"12\", \"read_rows\": \"1\"}")).build());

            Session insertSession = new Session()
                    .setSessionId("insert-session")
                    .setSessionCheck(false)
                    .setSessionTimeout(90)
                    .setSessionTimezone("America/Denver");
            InsertSettings insertSettings = new InsertSettings().use(insertSession);
            try (InsertResponse response = client.insert(
                    "test_table",
                    new ByteArrayInputStream("1\n".getBytes(StandardCharsets.UTF_8)),
                    ClickHouseFormat.CSV,
                    insertSettings).get(1, TimeUnit.SECONDS)) {
                Assert.assertEquals(response.getReadBytes(), 12);
            }
        } catch (Exception e) {
            Assert.fail("Unexpected exception", e);
        } finally {
            mockServer.stop();
        }
    }

    static {
        if (Boolean.getBoolean("test.debug")) {
            System.setProperty("org.slf4j.simpleLogger.defaultLogLevel", "DEBUG");
        }
    }

    @Test(groups = { "integration" })
    public void testSSLAuthentication() throws Exception {
        if (isCloud()) {
            return; // Current test is working only with local server because of self-signed certificates.
        }
        ClickHouseNode server = getSecureServer(ClickHouseProtocol.HTTP);
        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), true)
                .setUsername("dba")
                .setPassword("dba")
                .setRootCertificate("containers/clickhouse-server/certs/localhost.crt")
                .build()) {

            try (CommandResponse resp = client.execute("DROP USER IF EXISTS some_user").get()) {
            }
            try (CommandResponse resp = client.execute("CREATE USER some_user IDENTIFIED WITH ssl_certificate CN 'some_user'").get()) {
            }
        }

        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), true)
                .useSSLAuthentication(true)
                .setUsername("some_user")
                .setRootCertificate("containers/clickhouse-server/certs/localhost.crt")
                .setClientCertificate("some_user.crt")
                .setClientKey("some_user.key")
                .compressServerResponse(false)
                .build()) {

            try (QueryResponse resp = client.query("SELECT 1").get()) {
                Assert.assertEquals(resp.getReadRows(), 1);
            }
        }
    }

    @Test(groups = { "integration" }, dataProvider = "testPasswordAuthenticationProvider", dataProviderClass = HttpTransportTests.class)
    public void testPasswordAuthentication(String identifyWith, String identifyBy, boolean failsWithHeaders) throws Exception {
        if (isCloud()) {
            return; // Current test is working only with local server because of self-signed certificates.
        }
        ClickHouseNode server = getServer(ClickHouseProtocol.HTTP);

        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false)
                .setUsername("dba")
                .setPassword("dba")
                .build()) {

            try (CommandResponse resp = client.execute("DROP USER IF EXISTS some_user").get()) {
            }
            try (CommandResponse resp = client.execute("CREATE USER some_user IDENTIFIED WITH " + identifyWith + " BY '" + identifyBy + "'").get()) {
            }
        } catch (Exception e) {
            Assert.fail("Failed on setup", e);
        }


        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false)
                .setUsername("some_user")
                .setPassword(identifyBy)
                .build()) {

            Assert.assertEquals(client.queryAll("SELECT user()").get(0).getString(1), "some_user");
        } catch (Exception e) {
            Assert.fail("Failed to authenticate", e);
        }

        if (failsWithHeaders) {
            try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false)
                    .setUsername("some_user")
                    .setPassword(identifyBy)
                    .useHTTPBasicAuth(false)
                    .build()) {

                Assert.expectThrows(ClientException.class, () ->
                        client.queryAll("SELECT user()").get(0).getString(1));

            } catch (Exception e) {
                Assert.fail("Unexpected exception", e);
            }
        }
    }

    @DataProvider(name = "testPasswordAuthenticationProvider")
    public static Object[][] testPasswordAuthenticationProvider() {
        return new Object[][] {
                { "plaintext_password", "password", false},
                { "plaintext_password", "", false },
                { "plaintext_password", "S3Cr=?t", true},
                { "plaintext_password", "123§", true },
                { "sha256_password", "password", false },
                { "sha256_password", "123§", true },
                { "sha256_password", "S3Cr=?t", true},
                { "sha256_password", "S3Cr?=t", false},
        };
    }

    @Test(groups = { "integration" })
    public void testAuthHeaderIsKeptFromUser() throws Exception {
        if (isCloud()) {
            return; // Current test is working only with local server because of self-signed certificates.
        }
        ClickHouseNode server = getServer(ClickHouseProtocol.HTTP);

        String identifyWith = "sha256_password";
        String identifyBy = "123§";
        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false)
                .setUsername("dba")
                .setPassword("dba")
                .build()) {

            try (CommandResponse resp = client.execute("DROP USER IF EXISTS some_user").get()) {
            }
            try (CommandResponse resp = client.execute("CREATE USER some_user IDENTIFIED WITH " + identifyWith + " BY '" + identifyBy + "'").get()) {
            }
        } catch (Exception e) {
            Assert.fail("Failed on setup", e);
        }


        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false)
                .setUsername("some_user")
                .setPassword(identifyBy)
                .useHTTPBasicAuth(false) // disable basic auth to produce CH headers
                .httpHeader(HttpHeaders.AUTHORIZATION, "Basic " + Base64.getEncoder().encodeToString(("some_user:" +identifyBy).getBytes()))
                .build()) {

            Assert.assertEquals(client.queryAll("SELECT user()").get(0).getString(1), "some_user");
        } catch (Exception e) {
            Assert.fail("Failed to authenticate", e);
        }
    }

    @Test(groups = { "integration" })
    public void testSSLAuthentication_invalidConfig() throws Exception {
        if (isCloud()) {
            return; // Current test is working only with local server because of self-signed certificates.
        }
        ClickHouseNode server = getSecureServer(ClickHouseProtocol.HTTP);
        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), true)
                .useSSLAuthentication(true)
                .setUsername("some_user")
                .setPassword("s3cret")
                .setRootCertificate("containers/clickhouse-server/certs/localhost.crt")
                .setClientCertificate("some_user.crt")
                .setClientKey("some_user.key")
                .compressServerResponse(false)
                .build()) {
            fail("Expected exception");
        } catch (IllegalArgumentException e) {
            e.printStackTrace();
                Assert.assertTrue(e.getMessage().startsWith("Only one of password, access token or SSL authentication"));
        }
    }

    @Test(groups = { "integration" })
    public void testErrorWithSendProgressHeaders() throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        ClickHouseNode server = getServer(ClickHouseProtocol.HTTP);
        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost",server.getPort(), false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .build()) {

            try (CommandResponse resp = client.execute("DROP TABLE IF EXISTS test_omm_table").get()) {
            }
            try (CommandResponse resp = client.execute("CREATE TABLE test_omm_table ( val String) Engine = MergeTree ORDER BY () ").get()) {
            }

            QuerySettings settings = new QuerySettings()
                    .serverSetting("send_progress_in_http_headers", "1")
                    .serverSetting("max_memory_usage", "54M");

            try (QueryResponse resp = client.query("INSERT INTO test_omm_table SELECT randomString(16) FROM numbers(300000000)", settings).get()) {

            } catch (ServerException e) {
                // 241 - MEMORY_LIMIT_EXCEEDED or 243 -NOT_ENOUGH_SPACE
                Assert.assertTrue(e.getCode() == 241 || e.getCode() == 243);
            }
        }
    }


    @Test(groups = { "integration" }, dataProvider = "testUserAgentHasCompleteProductName_dataProvider", dataProviderClass = HttpTransportTests.class)
    public void testUserAgentHasCompleteProductName(String clientName, Pattern userAgentPattern) throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        ClickHouseNode server = getServer(ClickHouseProtocol.HTTP);
        try (Client client = new Client.Builder()
                .addEndpoint(server.getBaseUri())
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .setClientName(clientName)
                .build()) {

            String q1Id = UUID.randomUUID().toString();

            client.execute("SELECT 1", (CommandSettings) new CommandSettings().setQueryId(q1Id)).get().close();
            client.execute("SYSTEM FLUSH LOGS").get().close();

            List<GenericRecord> logRecords = client.queryAll("SELECT http_user_agent, http_referer, " +
                    " forwarded_for  FROM clusterAllReplicas('default', system.query_log) WHERE query_id = '" + q1Id + "'");
            Assert.assertFalse(logRecords.isEmpty(), "No records found in query log");

            for (GenericRecord record : logRecords) {
                System.out.println(record.getString("http_user_agent"));
                Assert.assertTrue(userAgentPattern.matcher(record.getString("http_user_agent")).matches(),
                        record.getString("http_user_agent") + " doesn't match \"" +
                                  userAgentPattern.pattern() + "\"");

            }
        }
    }


    @DataProvider(name = "testUserAgentHasCompleteProductName_dataProvider")
    public static Object[][] testUserAgentHasCompleteProductName_dataProvider() {
        return new Object[][] {
                { "", Pattern.compile("clickhouse-java-v2\\/.+ \\(.+\\) Apache-HttpClient\\/[\\d\\.]+$") },
                { "test-client/1.0", Pattern.compile("test-client/1.0 clickhouse-java-v2\\/.+ \\(.+\\) Apache-HttpClient\\/[\\d\\.]+$")},
                { "test-client/", Pattern.compile("test-client/ clickhouse-java-v2\\/.+ \\(.+\\) Apache-HttpClient\\/[\\d\\.]+$")}};
    }

    @Test(dataProvider = "testClientNameDataProvider")
    public void testClientName(String clientName, boolean setWithUpdate, String userAgentHeader, boolean setForRequest) throws Exception {

        final String initialClientName = setWithUpdate ? "init clientName" : clientName;
        final String initialUserAgentHeader = setForRequest ? "init userAgentHeader" : userAgentHeader;
        final String clientReferer = "http://localhost/webpage";

        Client.Builder builder = newClient();
        if (initialClientName != null) {
            builder.setClientName(initialClientName);
        }
        if (initialUserAgentHeader != null) {
            builder.httpHeader(HttpHeaders.USER_AGENT, initialUserAgentHeader);
        }
        try (Client client = builder.build()) {
            String expectedClientNameStartsWith = initialClientName == null || initialUserAgentHeader != null ? initialUserAgentHeader : initialClientName;

            if (setWithUpdate) {
                client.updateClientName(clientName);
                expectedClientNameStartsWith = initialUserAgentHeader == null ? clientName : initialUserAgentHeader;
            }

            String qId = UUID.randomUUID().toString();
            QuerySettings settings = new QuerySettings()
                    .httpHeader(HttpHeaders.REFERER, clientReferer)
                    .setQueryId(qId);

            if (setForRequest) {
                settings.httpHeader(HttpHeaders.USER_AGENT, userAgentHeader);
                expectedClientNameStartsWith = userAgentHeader;
            }

            client.query("SELECT 1", settings).get().close();
            client.execute("SYSTEM FLUSH LOGS").get().close();

            List<GenericRecord> logRecords = client.queryAll("SELECT query_id, client_name, http_user_agent, http_referer " +
                    " FROM clusterAllReplicas('default', system.query_log) WHERE query_id = '" + settings.getQueryId() + "'");
            Assert.assertEquals(logRecords.get(0).getString("query_id"), settings.getQueryId());
            final String logUserAgent = logRecords.get(0).getString("http_user_agent");
            Assert.assertTrue(logUserAgent.startsWith(expectedClientNameStartsWith),
                    "Expected to start with \"" + expectedClientNameStartsWith + "\" but values was \"" + logUserAgent + "\"" );
            Assert.assertTrue(logUserAgent.contains(Client.CLIENT_USER_AGENT), "Expected to contain client v2 version but value was \"" + logUserAgent + "\"");
            Assert.assertEquals(logRecords.get(0).getString("http_referer"), clientReferer);
            Assert.assertEquals(logRecords.get(0).getString("client_name"), ""); // http client can't set this field
        }
    }

    @DataProvider(name = "testClientNameDataProvider")
    public static Object[][] testClientName() {
        return new Object[][] {
                {"test-product (app 1.0)", false, null, false}, // only client name set
                {"test-product (app 1.0)", false, "final product (app 1.1)", false}, // http header set and overrides client name
                {"test-product (app 1.0)", true, null, false}, // client name set thru Client#updateClientName
                {"test-product (app 1.0)", true, "final product (app 1.1)", true}, // custom UserAgent header overrides client name
        };
    }

    @Test(dataProvider = "testClientNameThruRawOptionsDataProvider")
    public void testClientNameThruRawOptions(String property, String value, boolean setInClient) throws Exception {
        Client.Builder builder = newClient();
        if (setInClient) {
            builder.setOption(property, value);
        }
        try (Client client = builder.build()) {

            String qId = UUID.randomUUID().toString();
            QuerySettings settings = new QuerySettings()
                    .setQueryId(qId);

            if (!setInClient) {
                settings.setOption(property, value);
            }

            client.query("SELECT 1", settings).get().close();
            client.execute("SYSTEM FLUSH LOGS").get().close();

            List<GenericRecord> logRecords = client.queryAll("SELECT query_id, client_name, http_user_agent, http_referer " +
                    " FROM clusterAllReplicas('default', system.query_log) WHERE query_id = '" + settings.getQueryId() + "'");
            Assert.assertEquals(logRecords.get(0).getString("query_id"), settings.getQueryId());
            final String logUserAgent = logRecords.get(0).getString("http_user_agent");
            Assert.assertTrue(logUserAgent.startsWith(value),
                    "Expected to start with \"" + value + "\" but values was \"" + logUserAgent + "\"" );
            Assert.assertTrue(logUserAgent.contains(Client.CLIENT_USER_AGENT), "Expected to contain client v2 version but value was \"" + logUserAgent + "\"");
        }
    }

    @DataProvider(name = "testClientNameThruRawOptionsDataProvider")
    public Object[][] testClientNameThruRawOptionsDataProvider() {
        return new Object[][] {
                {ClientConfigProperties.PRODUCT_NAME.getKey(), "my product 1", true},
                {ClientConfigProperties.CLIENT_NAME.getKey(), "my product 2", true},
                {ClientConfigProperties.PRODUCT_NAME.getKey(), "my product 1", false},
                {ClientConfigProperties.CLIENT_NAME.getKey(), "my product 2", false},
        };
    }

    @Test(groups = { "integration" })
    public void testBearerTokenAuth() throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        WireMockServer mockServer = new WireMockServer( WireMockConfiguration
                .options().port(9090).notifier(new ConsoleNotifier(false)));
        mockServer.start();

        try {
            String jwtToken1 = Arrays.stream(
                            new String[]{"header", "payload", "signature"})
                    .map(s -> Base64.getEncoder().encodeToString(s.getBytes(StandardCharsets.UTF_8)))
                    .reduce((s1, s2) -> s1 + "." + s2).get();
            try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                    .useBearerTokenAuth(jwtToken1)
                    .compressServerResponse(false)
                    .build()) {

                mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                        .withHeader("Authorization", WireMock.equalTo("Bearer " + jwtToken1))
                        .willReturn(WireMock.aResponse()
                                .withHeader("X-ClickHouse-Summary",
                                        "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());

                try (QueryResponse response = client.query("SELECT 1").get(1, TimeUnit.SECONDS)) {
                    Assert.assertEquals(response.getReadBytes(), 10);
                } catch (Exception e) {
                    Assert.fail("Unexpected exception", e);
                }
            }

            String jwtToken2 = Arrays.stream(
                            new String[]{"header2", "payload2", "signature2"})
                    .map(s -> Base64.getEncoder().encodeToString(s.getBytes(StandardCharsets.UTF_8)))
                    .reduce((s1, s2) -> s1 + "." + s2).get();

            mockServer.resetAll();
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .withHeader("Authorization", WireMock.equalTo("Bearer " + jwtToken1))
                    .willReturn(WireMock.aResponse()
                            .withStatus(HttpStatus.SC_UNAUTHORIZED))
                    .build());

            try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                    .useBearerTokenAuth(jwtToken1)
                    .compressServerResponse(false)
                    .build()) {

                try {
                    client.execute("SELECT 1").get();
                    fail("Exception expected");
                } catch (ServerException e) {
                    Assert.assertEquals(e.getTransportProtocolCode(), HttpStatus.SC_UNAUTHORIZED);
                }

                mockServer.resetAll();
                mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                        .withHeader("Authorization", WireMock.equalTo("Bearer " + jwtToken2))
                        .willReturn(WireMock.aResponse()
                                .withHeader("X-ClickHouse-Summary",
                                        "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}"))

                        .build());

                client.updateBearerToken(jwtToken2);

                client.execute("SELECT 1").get();
            }
        } finally {
            mockServer.stop();
        }
    }

    @Test(groups = { "integration" })
    public void testUpdateSessionIdAtRuntime() throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().dynamicPort().notifier(new ConsoleNotifier(false)));
        mockServer.start();

        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .setSessionId("session-initial")
                .setSessionCheck(false)
                .setSessionTimeout(60)
                .setSessionTimezone("UTC")
                .compressServerResponse(false)
                .build()) {

            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .withQueryParam("session_id", WireMock.equalTo("session-initial"))
                    .withQueryParam("session_check", WireMock.equalTo("0"))
                    .withQueryParam("session_timeout", WireMock.equalTo("60"))
                    .withQueryParam("session_timezone", WireMock.equalTo("UTC"))
                    .willReturn(WireMock.aResponse()
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());

            try (CommandResponse response = client.execute("SELECT 1").get(1, TimeUnit.SECONDS)) {
                Assert.assertEquals(response.getReadBytes(), 10);
            }

            mockServer.resetAll();
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .withQueryParam("session_id", WireMock.equalTo("session-updated"))
                    .withQueryParam("session_check", WireMock.equalTo("0"))
                    .withQueryParam("session_timeout", WireMock.equalTo("60"))
                    .withQueryParam("session_timezone", WireMock.equalTo("UTC"))
                    .willReturn(WireMock.aResponse()
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"11\", \"read_rows\": \"1\"}")).build());

            client.updateSessionId("session-updated");

            try (CommandResponse response = client.execute("SELECT 1").get(1, TimeUnit.SECONDS)) {
                Assert.assertEquals(response.getReadBytes(), 11);
            }
        } finally {
            mockServer.stop();
        }
    }

    @Test(groups = { "integration" })
    public void testUpdateSessionIdRejectInvalidValues() {
        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", 8123, false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .build()) {
            Assert.assertThrows(IllegalArgumentException.class, () -> client.updateSessionId(""));
        }
    }

    @Test(groups = { "integration" })
    public void testBasicAuthWithNoPassword() throws Exception {
        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().port(9090).notifier(new ConsoleNotifier(false)));
        mockServer.start();

        try {
            // Expected: "default:" with empty password, not "default:null"
            String expectedAuth = "Basic " + Base64.getEncoder()
                    .encodeToString("default:".getBytes(StandardCharsets.UTF_8));

            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .withHeader("Authorization", WireMock.equalTo(expectedAuth))
                    .willReturn(WireMock.aResponse()
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());

            try (Client client = new Client.Builder()
                    .addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                    .compressServerResponse(false)
                    // no setPassword() call — password should default to empty, not "null"
                    .build()) {

                try (QueryResponse response = client.query("SELECT 1").get(1, TimeUnit.SECONDS)) {
                    Assert.assertEquals(response.getReadBytes(), 10);
                } catch (Exception e) {
                    Assert.fail("Basic auth with no password should send empty password, not 'null'", e);
                }
            }
        } finally {
            mockServer.stop();
        }
    }

    @Test(groups = { "integration" })
    public void testJWTWithCloud() throws Exception {
        if (!isCloud()) {
            return; // only for cloud
        }
        final String jwt = System.getenv("CLIENT_JWT");
        final String host = System.getenv("JWT_TEST_HOST");
        Assert.assertTrue(jwt != null && !jwt.trim().isEmpty(), "CLIENT_JWT is not set.");
        Assert.assertTrue(host != null && !host.trim().isEmpty(), "JWT_TEST_HOST is not set");
        Assert.assertFalse(jwt.contains("\n") || jwt.contains("-----"), "JWT should be single string ready for HTTP header");
        try (Client client = new Client.Builder()
                .addEndpoint(Protocol.HTTP, host, 8443, true)
                .setUsername("default")
                .compressClientRequest(false)
                .setDefaultDatabase("default")
                .serverSetting(ServerSettings.WAIT_END_OF_QUERY, "1")
                .useBearerTokenAuth(jwt).build()) {
            try {
                List<GenericRecord> response = client.queryAll("SELECT user(), now()");
                System.out.println("response: " + response.get(0).getString(1) + " time: " + response.get(0).getString(2));
            } catch (Exception e) {
                e.printStackTrace();
                throw e;
            }
        }
    }

    @Test(groups = { "integration" })
    public void testWithDefaultTimeouts() {
        if (isCloud()) {
            return; // mocked server
        }

        int proxyPort = new Random().nextInt(1000) + 10000;
        WireMockServer proxy = new WireMockServer(WireMockConfiguration
                .options().port(proxyPort)
                .notifier(new Slf4jNotifier(true)));
        proxy.start();
        proxy.addStubMapping(WireMock.post(WireMock.anyUrl())
                .willReturn(WireMock.aResponse().withFixedDelay(5000)
                        .withStatus(HttpStatus.SC_OK)
                        .withHeader("X-ClickHouse-Summary", "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());

        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", proxyPort, false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .build()) {
            int startTime = (int) System.currentTimeMillis();
            try {
                client.query("SELECT 1").get();
            } catch (Exception e) {
                Assert.fail("Elapsed Time: " + (System.currentTimeMillis() - startTime), e);
            }
        } finally {
            proxy.stop();
        }
    }


    @Test(groups = { "integration" })
    public void testTimeoutsWithRetry() {
        if (isCloud()) {
            return; // mocked server
        }

        WireMockServer faultyServer = new WireMockServer( WireMockConfiguration
                .options().port(9090).notifier(new ConsoleNotifier(false)));
        faultyServer.start();

        // First request gets no response
        faultyServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                .inScenario("Timeout")
                .withRequestBody(WireMock.containing("SELECT 1"))
                .whenScenarioStateIs(STARTED)
                .willSetStateTo("Failed")
                .willReturn(WireMock.aResponse()
                        .withStatus(HttpStatus.SC_OK)
                        .withFixedDelay(5000)
                        .withHeader("X-ClickHouse-Summary",
                        "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());

        // Second request gets a response (retry)
        faultyServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                .inScenario("Timeout")
                .withRequestBody(WireMock.containing("SELECT 1"))
                .whenScenarioStateIs("Failed")
                .willSetStateTo("Done")
                .willReturn(WireMock.aResponse()
                        .withStatus(HttpStatus.SC_OK)
                        .withFixedDelay(1000)
                        .withHeader("X-ClickHouse-Summary",
                                "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());

        try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", faultyServer.port(), false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .setSocketTimeout(3000)
                .retryOnFailures(ClientFaultCause.SocketTimeout)
                .build()) {
            int startTime = (int) System.currentTimeMillis();
            try {
                client.query("SELECT 1").get();
            } catch (Exception e) {
                Assert.fail("Elapsed Time: " + (System.currentTimeMillis() - startTime), e);
            }
        } finally {
            faultyServer.stop();
        }
    }

    @Test(groups = {"integration"})
    public void testSNIWithCloud() throws Exception {
        if (!isCloud()) {
            // skip for local env
            return;
        }

        ClickHouseNode node = getServer(ClickHouseProtocol.HTTP);
        String ip = InetAddress.getByName(node.getHost()).getHostAddress();
        try (Client c = new Client.Builder()
                .addEndpoint(Protocol.HTTP, ip, node.getPort(), true)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .sslSocketSNI(node.getHost()).build()) {
            c.execute("SELECT 1");
        }
    }

    @Test(groups = {"integration"})
    public void testEndpointUrlPathIsPreserved() throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        int serverPort = new Random().nextInt(1000) + 10000;
        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().port(serverPort)
                .notifier(new Slf4jNotifier(true)));
        mockServer.start();

        try {
            // Setup stubs for two virtual ClickHouse instances behind a reverse proxy
            mockServer.addStubMapping(WireMock.post(WireMock.urlPathEqualTo("/sales/db"))
                    .willReturn(WireMock.aResponse()
                            .withStatus(HttpStatus.SC_OK)
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"100\", \"read_rows\": \"10\"}")).build());

            mockServer.addStubMapping(WireMock.post(WireMock.urlPathEqualTo("/billing/db"))
                    .willReturn(WireMock.aResponse()
                            .withStatus(HttpStatus.SC_OK)
                            .withHeader("X-ClickHouse-Summary",
                                    "{ \"read_bytes\": \"200\", \"read_rows\": \"20\"}")).build());

            // Test sales virtual instance
            try (Client salesClient = new Client.Builder()
                    .addEndpoint("http://localhost:" + serverPort + "/sales/db")
                    .setUsername("default")
                    .setPassword(ClickHouseServerForTest.getPassword())
                    .compressServerResponse(false)
                    .build()) {

                try (QueryResponse response = salesClient.query("SELECT 1").get(10, TimeUnit.SECONDS)) {
                    Assert.assertEquals(response.getReadBytes(), 100);
                }
            }

            // Test billing virtual instance - also verify query parameters in URL are ignored
            try (Client billingClient = new Client.Builder()
                    .addEndpoint("http://localhost:" + serverPort + "/billing/db?ignored_param=value")
                    .setUsername("default")
                    .setPassword(ClickHouseServerForTest.getPassword())
                    .compressServerResponse(false)
                    .build()) {

                try (QueryResponse response = billingClient.query("SELECT 1").get(10, TimeUnit.SECONDS)) {
                    Assert.assertEquals(response.getReadBytes(), 200);
                }

                // Verify that ignored_param is not in the request URL
                mockServer.verify(WireMock.postRequestedFor(WireMock.urlPathEqualTo("/billing/db"))
                        .withoutQueryParam("ignored_param"));
            }

            // Verify requests were made to the correct paths
            mockServer.verify(WireMock.postRequestedFor(WireMock.urlPathEqualTo("/sales/db")));
            mockServer.verify(WireMock.postRequestedFor(WireMock.urlPathEqualTo("/billing/db")));

        } finally {
            mockServer.stop();
        }
    }

    @Test(groups = {"integration"})
    public void testMultiPartRequest() {
        final Map<String, Object> params = new HashMap<>();
        params.put("database_name", "system");
        params.put("table_names",
                DataTypeConverter.INSTANCE.arrayToString(Arrays.asList("COLLATIONS", "ENGINES"), "Array(String)"));

        // Use http compression
        try (Client client = newClient().useHttpCompression(true).setOption(ClientConfigProperties.HTTP_SEND_PARAMS_IN_BODY.getKey(), "true").build()) {
            List<GenericRecord> records = client.queryAll("SELECT database, name FROM system.tables WHERE name IN {table_names:Array(String)}",
                    params);

            Assert.assertEquals(records.get(0).getString("name"), "COLLATIONS");
            Assert.assertEquals(records.get(1).getString("name"), "ENGINES");
        }

        // Use http compression
        try (Client client = newClient().useHttpCompression(false).setOption(ClientConfigProperties.HTTP_SEND_PARAMS_IN_BODY.getKey(), "true").build()) {
            List<GenericRecord> records = client.queryAll("SELECT database, name FROM system.tables WHERE name IN {table_names:Array(String)}",
                    params);

            Assert.assertEquals(records.get(0).getString("name"), "COLLATIONS");
            Assert.assertEquals(records.get(1).getString("name"), "ENGINES");
        }

        // compress request
        try (Client client = newClient()
                .compressClientRequest(true)
                .setOption(ClientConfigProperties.HTTP_SEND_PARAMS_IN_BODY.getKey(), "true")
                .useHttpCompression(true).build()) {
            List<GenericRecord> records = client.queryAll("SELECT database, name FROM system.tables WHERE name IN {table_names:Array(String)}",
                    params);

            Assert.assertEquals(records.get(0).getString("name"), "COLLATIONS");
            Assert.assertEquals(records.get(1).getString("name"), "ENGINES");
        }
    }

    @Test(groups = {"integration"})
    public void testSmallNetworkBufferDoesNotBreakColumnDecoding() throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        final int rowsToRead = 2_000;
        final int networkBufferSize = 8196;
        final String query = "SELECT toUInt32(number), toUInt64(number), now() FROM system.numbers LIMIT " + rowsToRead;
        byte[] validBody = fetchBinaryPayload(query, networkBufferSize, 60);

        Assert.assertTrue(validBody.length > 3, "Source binary payload is unexpectedly small");
        byte[] corruptedBody = Arrays.copyOf(validBody, validBody.length - 5);

        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().dynamicPort().notifier(new ConsoleNotifier(false)));
        mockServer.start();
        try {
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .willReturn(WireMock.aResponse()
                            .withStatus(HttpStatus.SC_OK)
                            .withHeader("X-ClickHouse-Summary", "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")
                            .withBody(corruptedBody))
                    .build());
            Throwable thrown = assertBinaryDecodeFails(mockServer, query, networkBufferSize, 60,
                    "Expected failure when reading truncated binary stream");
            assertBinaryReadFailureContainsColumnName(thrown);
        } finally {
            mockServer.stop();
        }
    }

    @Test(groups = {"integration"})
    public void testChunkedResponsePrematureEndIsReported() throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        final String query = "SELECT toUInt32(number), toUInt64(number), now() FROM system.numbers LIMIT 10";

        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().dynamicPort().notifier(new ConsoleNotifier(false)));
        mockServer.start();
        try {
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .willReturn(WireMock.aResponse()
                            .withStatus(HttpStatus.SC_OK)
                            .withHeader(HttpHeaders.TRANSFER_ENCODING, "chunked")
                            .withFault(Fault.MALFORMED_RESPONSE_CHUNK))
                    .build());
            Throwable thrown = assertBinaryDecodeFails(mockServer, query, null, 30,
                    "Expected failure when reading malformed chunked response");
            ConnectionClosedException connectionClosedException = findCause(thrown, ConnectionClosedException.class);
            boolean hasChunkedPrematureCloseSignature = containsMessageInCauseChain(thrown,
                    "closing chunk expected",
                    "premature end of chunk coded message body",
                    "failed to read header");
            Assert.assertTrue(connectionClosedException != null || hasChunkedPrematureCloseSignature,
                    "Expected chunked/premature-close failure signature, but was: " + thrown);
        } finally {
            mockServer.stop();
        }
    }

    @Test(groups = {"integration"})
    public void testTailCorruptedStreamFailsDecoding() throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        final int rowsToRead = 100_000;
        final int networkBufferSize = 8196;
        final String query = "SELECT toUInt32(number), toUInt64(number), now() FROM system.numbers LIMIT " + rowsToRead;
        byte[] validBody = fetchBinaryPayload(query, networkBufferSize, 60);

        final int removedBytes = 3;
        Assert.assertTrue(validBody.length > removedBytes, "Source binary payload is unexpectedly small");
        byte[] corruptedBody = Arrays.copyOf(validBody, validBody.length - removedBytes);

        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().dynamicPort().notifier(new ConsoleNotifier(false)));
        mockServer.start();
        try {
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .willReturn(WireMock.aResponse()
                            .withStatus(HttpStatus.SC_OK)
                            .withHeader("X-ClickHouse-Summary", "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")
                            .withBody(corruptedBody))
                    .build());
            Throwable thrown = assertBinaryDecodeFails(mockServer, query, networkBufferSize, 60,
                    "Expected failure when reading binary stream truncated at tail");
            assertBinaryReadFailureContainsColumnName(thrown);
        } finally {
            mockServer.stop();
        }
    }

    @Test(groups = {"integration"})
    public void testTailStreamFailureReportsPositiveTimeSinceLastNextCall() throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        final int rowsToRead = 2000;
        final int networkBufferSize = 8196;
        final String query = "SELECT toUInt32(number), toUInt64(number), now() FROM system.numbers LIMIT " + rowsToRead;
        byte[] validBody = fetchBinaryPayload(query, networkBufferSize, 60);
        final int removedBytes = 3;
        Assert.assertTrue(validBody.length > removedBytes, "Source binary payload is unexpectedly small");
        byte[] corruptedBody = Arrays.copyOf(validBody, validBody.length - removedBytes);

        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().dynamicPort().notifier(new ConsoleNotifier(false)));
        mockServer.start();
        try {
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .willReturn(WireMock.aResponse()
                            .withStatus(HttpStatus.SC_OK)
                            .withHeader("X-ClickHouse-Summary", "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")
                            .withBody(corruptedBody))
                    .build());

            QuerySettings querySettings = new QuerySettings().setFormat(ClickHouseFormat.RowBinaryWithNamesAndTypes);
            try (Client client = newMockServerClient(mockServer.port(), networkBufferSize)
                    .build();
                 QueryResponse response = client.query(query, querySettings).get(60, TimeUnit.SECONDS);
                 ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(response)) {
                final int[] rowsRead = new int[] {0};
                Throwable thrown = Assert.expectThrows(Throwable.class, () -> {
                    while (true) {
                        if (rowsRead[0] >= 5) {
                            try {
                                Thread.sleep(25);
                            } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                                throw new RuntimeException(e);
                            }
                        }
                        if (reader.next() == null) {
                            return;
                        }
                        rowsRead[0]++;
                    }
                });

                Assert.assertTrue(rowsRead[0] >= 5,
                        "Expected to read at least a few rows before failure, but read " + rowsRead[0]);
                ClientException clientException = findCause(thrown, ClientException.class);
                Assert.assertNotNull(clientException,
                        "Expected ClientException in cause chain, but was: " + thrown);
                Assert.assertTrue(containsMessageInCauseChain(thrown, "Reading column "),
                        "Expected column information in failure message chain, but was: " + thrown);

                String elapsedTimeMessage = findFirstMessageInCauseChain(thrown, "time since last next call");
                Assert.assertNotNull(elapsedTimeMessage,
                        "Expected elapsed-time fragment in failure message chain, but was: " + thrown);

                java.util.regex.Matcher matcher = Pattern.compile("time since last next call (\\d+)\\)")
                        .matcher(elapsedTimeMessage);
                Assert.assertTrue(matcher.find(),
                        "Expected elapsed-time fragment in message: " + elapsedTimeMessage);
                long elapsedSinceLastNext = Long.parseLong(matcher.group(1));
                Assert.assertTrue(elapsedSinceLastNext > 0,
                        "Expected positive elapsed time since last next call, but was " + elapsedSinceLastNext);
            }
        } finally {
            mockServer.stop();
        }
    }

    @Test(groups = {"integration"}, dataProvider = "testHttpStatusErrorBodyDataProvider")
    public void testHttpStatusErrorsIncludeResponseBody(int httpStatus, String responseBody, String expectedBodyPart) throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().dynamicPort().notifier(new ConsoleNotifier(false)));
        mockServer.start();

        try {
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .willReturn(WireMock.aResponse()
                            .withStatus(httpStatus)
                            .withBody(responseBody))
                    .build());

            try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                    .setUsername("default")
                    .setPassword(ClickHouseServerForTest.getPassword())
                    .compressServerResponse(false)
                    .build()) {

                Throwable thrown = Assert.expectThrows(Throwable.class,
                        () -> client.query("SELECT 1").get(1, TimeUnit.SECONDS));
                ServerException serverException = findServerException(thrown);
                Assert.assertNotNull(serverException, "Expected ServerException in cause chain");
                Assert.assertEquals(serverException.getTransportProtocolCode(), httpStatus);
                Assert.assertTrue(serverException.getMessage().contains(expectedBodyPart),
                        "Expected to contain '" + expectedBodyPart + "', but was: " + serverException.getMessage());
            }
        } finally {
            mockServer.stop();
        }
    }

    @DataProvider(name = "testHttpStatusErrorBodyDataProvider")
    public static Object[][] testHttpStatusErrorBodyDataProvider() {
        return new Object[][]{
                {HttpStatus.SC_UNAUTHORIZED, "Unauthorized: invalid credentials for user default", "invalid credentials"},
                {HttpStatus.SC_FORBIDDEN, "Forbidden: user default has no access to this operation", "no access"},
                {HttpStatus.SC_NOT_FOUND, "Not found: requested endpoint does not exist", "requested endpoint"}
        };
    }

    @Test(groups = {"integration"}, dataProvider = "testHttpStatusWithoutBodyDataProvider")
    public void testHttpStatusErrorsWithoutBody(int httpStatus) throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().dynamicPort().notifier(new ConsoleNotifier(false)));
        mockServer.start();

        try {
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .willReturn(WireMock.aResponse().withStatus(httpStatus))
                    .build());

            try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                    .setUsername("default")
                    .setPassword(ClickHouseServerForTest.getPassword())
                    .compressServerResponse(false)
                    .build()) {

                Throwable thrown = Assert.expectThrows(Throwable.class,
                        () -> client.query("SELECT 1").get(1, TimeUnit.SECONDS));
                ServerException serverException = findServerException(thrown);
                Assert.assertNotNull(serverException, "Expected ServerException in cause chain");
                Assert.assertEquals(serverException.getTransportProtocolCode(), httpStatus);
                Assert.assertTrue(serverException.getMessage().contains("unknown server error"),
                        "Expected unknown error message for empty body, but was: " + serverException.getMessage());
            }
        } finally {
            mockServer.stop();
        }
    }

    @DataProvider(name = "testHttpStatusWithoutBodyDataProvider")
    public static Object[][] testHttpStatusWithoutBodyDataProvider() {
        return new Object[][]{
                {HttpStatus.SC_UNAUTHORIZED},
                {HttpStatus.SC_FORBIDDEN},
                {HttpStatus.SC_NOT_FOUND}
        };
    }

    @Test(groups = {"integration"}, dataProvider = "testHttpStatusCompressedBodyDataProvider")
    public void testHttpStatusErrorsWithHttpCompression(int httpStatus, String responseBody, String expectedBodyPart) throws Exception {
        if (isCloud()) {
            return; // mocked server
        }

        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
                .options().dynamicPort().notifier(new ConsoleNotifier(false)));
        mockServer.start();

        try {
            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
                    .willReturn(WireMock.aResponse()
                            .withStatus(httpStatus)
                            .withHeader(HttpHeaders.CONTENT_ENCODING, "gzip")
                            .withBody(gzip(responseBody)))
                    .build());

            try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
                    .setUsername("default")
                    .setPassword(ClickHouseServerForTest.getPassword())
                    .useHttpCompression(true)
                    .compressServerResponse(true)
                    .build()) {

                Throwable thrown = Assert.expectThrows(Throwable.class,
                        () -> client.query("SELECT 1").get(1, TimeUnit.SECONDS));
                ServerException serverException = findServerException(thrown);
                Assert.assertNotNull(serverException, "Expected ServerException in cause chain");
                Assert.assertEquals(serverException.getTransportProtocolCode(), httpStatus);
                Assert.assertTrue(serverException.getMessage().contains(expectedBodyPart),
                        "Expected compressed body part '" + expectedBodyPart + "', but was: " + serverException.getMessage());
            }
        } finally {
            mockServer.stop();
        }
    }

    @DataProvider(name = "testHttpStatusCompressedBodyDataProvider")
    public static Object[][] testHttpStatusCompressedBodyDataProvider() {
        return new Object[][]{
                {HttpStatus.SC_UNAUTHORIZED, "Unauthorized: token is expired", "token is expired"},
                {HttpStatus.SC_FORBIDDEN, "Forbidden: policy denies this query", "policy denies"},
                {HttpStatus.SC_NOT_FOUND, "Not found: route does not exist", "route does not exist"}
        };
    }

    private byte[] fetchBinaryPayload(String query, int networkBufferSize, int timeoutSec) throws Exception {
        QuerySettings querySettings = new QuerySettings().setFormat(ClickHouseFormat.RowBinaryWithNamesAndTypes);
        try (Client client = newClient()
                .useHttpCompression(false)
                .compressServerResponse(false)
                .setOption(ClientConfigProperties.CLIENT_NETWORK_BUFFER_SIZE.getKey(), String.valueOf(networkBufferSize))
                .build();
             QueryResponse response = client.query(query, querySettings).get(timeoutSec, TimeUnit.SECONDS)) {
            return readAllBytes(response.getInputStream());
        }
    }

    private Throwable assertBinaryDecodeFails(WireMockServer mockServer, String query, Integer networkBufferSize,
                                              int timeoutSec, String assertionMessage) throws Exception {
        QuerySettings querySettings = new QuerySettings().setFormat(ClickHouseFormat.RowBinaryWithNamesAndTypes);
        try (Client client = newMockServerClient(mockServer.port(), networkBufferSize).build()) {
            Throwable thrown = Assert.expectThrows(Throwable.class, () -> {
                try (QueryResponse response = client.query(query, querySettings).get(timeoutSec, TimeUnit.SECONDS);
                     ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(response)) {
                    readAllRows(reader);
                }
            });
            Assert.assertNotNull(thrown, assertionMessage);
            return thrown;
        }
    }

    private Client.Builder newMockServerClient(int port, Integer networkBufferSize) {
        Client.Builder builder = new Client.Builder()
                .addEndpoint(Protocol.HTTP, "localhost", port, false)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .setDefaultDatabase(ClickHouseServerForTest.getDatabase())
                .serverSetting(ServerSettings.WAIT_END_OF_QUERY, "1")
                .useHttpCompression(false)
                .compressServerResponse(false);
        if (networkBufferSize != null) {
            builder.setOption(ClientConfigProperties.CLIENT_NETWORK_BUFFER_SIZE.getKey(), String.valueOf(networkBufferSize));
        }
        return builder;
    }

    private static void readAllRows(ClickHouseBinaryFormatReader reader) {
        while (reader.next() != null) {
            reader.getInteger(1);
            reader.getLong(2);
            reader.getString(3);
        }
    }

    private static byte[] gzip(String value) throws Exception {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        try (GZIPOutputStream gzipOutputStream = new GZIPOutputStream(out)) {
            gzipOutputStream.write(value.getBytes(StandardCharsets.UTF_8));
        }
        return out.toByteArray();
    }

    private static byte[] readAllBytes(java.io.InputStream inputStream) throws Exception {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        byte[] buffer = new byte[8192];
        int read;
        while ((read = inputStream.read(buffer)) != -1) {
            out.write(buffer, 0, read);
        }
        return out.toByteArray();
    }

    private static ServerException findServerException(Throwable throwable) {
        Throwable current = throwable;
        while (current != null) {
            if (current instanceof ServerException) {
                return (ServerException) current;
            }
            current = current.getCause();
        }
        return null;
    }

    private static <T extends Throwable> T findCause(Throwable throwable, Class<T> clazz) {
        Throwable current = throwable;
        while (current != null) {
            if (clazz.isInstance(current)) {
                return clazz.cast(current);
            }
            current = current.getCause();
        }
        return null;
    }

    private static boolean containsMessageInCauseChain(Throwable throwable, String... parts) {
        Throwable current = throwable;
        while (current != null) {
            String message = current.getMessage();
            if (message != null) {
                String lower = message.toLowerCase();
                for (String part : parts) {
                    if (lower.contains(part.toLowerCase())) {
                        return true;
                    }
                }
            }
            current = current.getCause();
        }
        return false;
    }

    private static String findFirstMessageInCauseChain(Throwable throwable, String part) {
        Throwable current = throwable;
        while (current != null) {
            String message = current.getMessage();
            if (message != null && message.contains(part)) {
                return message;
            }
            current = current.getCause();
        }
        return null;
    }

    private static void assertBinaryReadFailureContainsColumnName(Throwable thrown) {
        ClientException clientException = findCause(thrown, ClientException.class);
        Assert.assertNotNull(clientException,
                "Expected ClientException in cause chain, but was: " + thrown);
        Assert.assertTrue(containsMessageInCauseChain(thrown, "Reading column "),
                "Expected column information in failure message chain, but was: " + thrown);
    }

    protected Client.Builder newClient() {
        ClickHouseNode node = getServer(ClickHouseProtocol.HTTP);
        boolean isSecure = isCloud();
        return new Client.Builder()
                .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isSecure)
                .setUsername("default")
                .setPassword(ClickHouseServerForTest.getPassword())
                .compressClientRequest(false)
                .setDefaultDatabase(ClickHouseServerForTest.getDatabase())
                .serverSetting(ServerSettings.WAIT_END_OF_QUERY, "1");
    }
}