Merge pull request #2000 from ClickHouse/v2_connection_problems_investigation

chernser · web-flow · commit 7be8d90f5c0f · 2024-12-06T19:06:13.000-08:00
[client-v2] Fix ClickHouseLZ4OutputStream sending extra bytes when income buffer is empty
diff --git a/client-v2/src/main/java/com/clickhouse/client/api/Client.java b/client-v2/src/main/java/com/clickhouse/client/api/Client.java
@@ -1501,7 +1501,7 @@ public CompletableFuture<InsertResponse> insert(String tableName,
 
                                              byte[] buffer = new byte[writeBufferSize];
                                              int bytesRead;
-                                             while ((bytesRead = data.read(buffer)) != -1) {
+                                             while ((bytesRead = data.read(buffer)) > 0) {
                                                  out.write(buffer, 0, bytesRead);
                                              }
                                              out.close();
diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/ClickHouseLZ4OutputStream.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/ClickHouseLZ4OutputStream.java
@@ -65,21 +65,23 @@ public void write( byte[] b, int off, int len) throws IOException {
 
     @Override
     public void flush() throws IOException {
-        compressedBuffer.clear();
-        compressedBuffer.put(16, ClickHouseLZ4InputStream.MAGIC);
-        int uncompressedLen = buffer.position();
-        buffer.flip();
-        int compressed = compressor.compress(buffer, 0, uncompressedLen, compressedBuffer, 25,
-                compressedBuffer.remaining() - 25);
-        int compressedSizeWithHeader = compressed + 9;
-        ClickHouseLZ4InputStream.setInt32(compressedBuffer.array(), 17, compressedSizeWithHeader); // compressed size with header
-        ClickHouseLZ4InputStream.setInt32(compressedBuffer.array(), 21, uncompressedLen); // uncompressed size
-        long[] hash = ClickHouseCityHash.cityHash128(compressedBuffer.array(), 16, compressedSizeWithHeader);
-        setInt64(compressedBuffer.array(), 0, hash[0]);
-        setInt64(compressedBuffer.array(), 8, hash[1]);
-        compressedBuffer.flip();
-        out.write(compressedBuffer.array(), 0, compressed + 25);
-        buffer.clear();
+        if (buffer.position() > 0) {
+            compressedBuffer.clear();
+            compressedBuffer.put(16, ClickHouseLZ4InputStream.MAGIC);
+            int uncompressedLen = buffer.position();
+            buffer.flip();
+            int compressed = compressor.compress(buffer, 0, uncompressedLen, compressedBuffer, 25,
+                    compressedBuffer.remaining() - 25);
+            int compressedSizeWithHeader = compressed + 9;
+            ClickHouseLZ4InputStream.setInt32(compressedBuffer.array(), 17, compressedSizeWithHeader); // compressed size with header
+            ClickHouseLZ4InputStream.setInt32(compressedBuffer.array(), 21, uncompressedLen); // uncompressed size
+            long[] hash = ClickHouseCityHash.cityHash128(compressedBuffer.array(), 16, compressedSizeWithHeader);
+            setInt64(compressedBuffer.array(), 0, hash[0]);
+            setInt64(compressedBuffer.array(), 8, hash[1]);
+            compressedBuffer.flip();
+            out.write(compressedBuffer.array(), 0, compressed + 25);
+            buffer.clear();
+        }
     }
 
 
diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java
@@ -73,6 +73,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
+import static com.clickhouse.client.api.ClientConfigProperties.SOCKET_TCP_NO_DELAY_OPT;
+
 public class HttpAPIClientHelper {
     private static final Logger LOG = LoggerFactory.getLogger(Client.class);
 
@@ -235,6 +237,9 @@ public CloseableHttpClient createHttpClient() {
                 soCfgBuilder::setSndBufSize);
         MapUtils.applyInt(chConfiguration, ClientConfigProperties.SOCKET_LINGER_OPT.getKey(),
                     (v) -> soCfgBuilder.setSoLinger(v, TimeUnit.SECONDS));
+        if (MapUtils.getFlag(chConfiguration, ClientConfigProperties.SOCKET_TCP_NO_DELAY_OPT.getKey(), false)) {
+            soCfgBuilder.setTcpNoDelay(true);
+        }
 
         // Proxy
         String proxyHost = chConfiguration.get(ClientConfigProperties.PROXY_HOST.getKey());
diff --git a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java
@@ -223,7 +223,7 @@ public void testSecureConnection() {
                 .setUsername("default")
                 .setPassword("")
                 .setRootCertificate("containers/clickhouse-server/certs/localhost.crt")
-                .useNewImplementation(System.getProperty("client.tests.useNewImplementation", "true").equals("true"))
+                .compressClientRequest(true)
                 .build()) {
 
             List<GenericRecord> records = client.queryAll("SELECT timezone()");
diff --git a/client-v2/src/test/java/com/clickhouse/client/insert/InsertClientContentCompressionTests.java b/client-v2/src/test/java/com/clickhouse/client/insert/InsertClientContentCompressionTests.java
@@ -1,7 +1,77 @@
 package com.clickhouse.client.insert;
 
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.api.Client;
+import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader;
+import com.clickhouse.client.api.insert.InsertResponse;
+import com.clickhouse.client.api.insert.InsertSettings;
+import com.clickhouse.client.api.query.GenericRecord;
+import com.clickhouse.client.api.query.QueryResponse;
+import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
 public class InsertClientContentCompressionTests extends InsertTests {
     public InsertClientContentCompressionTests() {
         super(true, false);
     }
+
+
+    @Test(groups = { "integration" })
+    public void testInsertAndReadBackWithSecureConnection() {
+        ClickHouseNode secureServer = getSecureServer(ClickHouseProtocol.HTTP);
+
+        try (Client client = new Client.Builder()
+                .addEndpoint("https://localhost:" + secureServer.getPort())
+                .setUsername("default")
+                .setPassword("")
+                .setRootCertificate("containers/clickhouse-server/certs/localhost.crt")
+                .compressClientRequest(true)
+                .build()) {
+            final String tableName = "single_pojo_table";
+            final String createSQL = SamplePOJO.generateTableCreateSQL(tableName);
+            final SamplePOJO pojo = new SamplePOJO();
+
+            initTable(tableName, createSQL);
+
+            client.register(SamplePOJO.class, client.getTableSchema(tableName, "default"));
+            InsertSettings settings = new InsertSettings()
+                    .setDeduplicationToken(RandomStringUtils.randomAlphabetic(36))
+                    .setQueryId(String.valueOf(UUID.randomUUID()));
+            System.out.println("Inserting POJO: " + pojo);
+            try (InsertResponse response = client.insert(tableName, Collections.singletonList(pojo), settings).get(10, TimeUnit.SECONDS)) {
+                Assert.assertEquals(response.getWrittenRows(), 1);
+            }
+
+            try (QueryResponse queryResponse =
+                         client.query("SELECT * FROM " + tableName + " LIMIT 1").get(10, TimeUnit.SECONDS)) {
+
+                ClickHouseBinaryFormatReader reader = client.newBinaryFormatReader(queryResponse);
+                Assert.assertNotNull(reader.next());
+
+                Assert.assertEquals(reader.getByte("byteValue"), pojo.getByteValue());
+                Assert.assertEquals(reader.getByte("int8"), pojo.getInt8());
+                Assert.assertEquals(reader.getShort("uint8"), pojo.getUint8());
+                Assert.assertEquals(reader.getShort("int16"), pojo.getInt16());
+                Assert.assertEquals(reader.getInteger("int32"), pojo.getInt32());
+                Assert.assertEquals(reader.getLong("int64"), pojo.getInt64());
+                Assert.assertEquals(reader.getFloat("float32"), pojo.getFloat32());
+                Assert.assertEquals(reader.getDouble("float64"), pojo.getFloat64());
+                Assert.assertEquals(reader.getString("string"), pojo.getString());
+                Assert.assertEquals(reader.getString("fixedString"), pojo.getFixedString());
+            }
+            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());
+        }
+    }
 }
diff --git a/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java b/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java
@@ -66,14 +66,18 @@ public InsertTests(boolean useClientCompression, boolean useHttpCompression) {
     @BeforeMethod(groups = { "integration" })
     public void setUp() throws IOException {
         ClickHouseNode node = getServer(ClickHouseProtocol.HTTP);
+        int bufferSize = (7 * 65500);
         client = new Client.Builder()
                 .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), false)
                 .setUsername("default")
                 .setPassword("")
-                .useNewImplementation(System.getProperty("client.tests.useNewImplementation", "true").equals("true"))
                 .compressClientRequest(useClientCompression)
                 .useHttpCompression(useHttpCompression)
+                .setSocketSndbuf(bufferSize)
+                .setSocketRcvbuf(bufferSize)
+                .setClientNetworkBufferSize(bufferSize)
                 .build();
+
         settings = new InsertSettings()
                 .setDeduplicationToken(RandomStringUtils.randomAlphabetic(36))
                 .setQueryId(String.valueOf(UUID.randomUUID()));
@@ -226,7 +230,6 @@ public void insertRawData() throws Exception {
         assertEquals(records.size(), 1000);
     }
 
-
     @Test(groups = { "integration" }, enabled = true)
     public void insertRawDataSimple() throws Exception {
         insertRawDataSimple(1000);
@@ -379,7 +382,7 @@ public static Object[] logCommentDataProvider() {
         };
     }
 
-    private void initTable(String tableName, String createTableSQL) throws Exception {
+    protected void initTable(String tableName, String createTableSQL) throws Exception {
         client.execute("DROP TABLE IF EXISTS " + tableName).get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS);
         client.execute(createTableSQL).get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS);
     }

Original file line number	Diff line number	Diff line change
`@@ -1501,7 +1501,7 @@ public CompletableFuture<InsertResponse> insert(String tableName,`
`1501`	`1501`
`1502`	`1502`	`byte[] buffer = new byte[writeBufferSize];`
`1503`	`1503`	`int bytesRead;`
`1504`		`- while ((bytesRead = data.read(buffer)) != -1) {`
	`1504`	`+ while ((bytesRead = data.read(buffer)) > 0) {`
`1505`	`1505`	`out.write(buffer, 0, bytesRead);`
`1506`	`1506`	`}`
`1507`	`1507`	`out.close();`