Merge pull request #796 from ClickHouse/07/17/26/fix_writing_with_v2

[writing data] RowBinaryInsert with Client V2

Author: chernser

CHANGELOG.md

@@ -1,3 +1,15 @@
+# 1.5.0 (not published)
+
+## Improvements
+* RowBinary inserts with client V2 now stream data directly to the network output stream via the client's
+  `DataStreamWriter` API instead of first serializing the whole batch into an unsized `ByteArrayOutputStream`
+  (which caused repeated internal array reallocations and copies) and then copying it again through a
+  `ByteArrayInputStream`. Serialization is deterministic, so client-level retries resend a byte-identical
+  block and ClickHouse block deduplication (and therefore exactly-once delivery) is unaffected.
+  The record serialization loop is now shared between the V1 and V2 insert paths. Since client V2 transmits
+  the INSERT statement as an HTTP query parameter, feature test coverage was added for table names containing
+  URL-special characters (space, `+`, `&`, `=`, `%`, `?`, `#`), SQL quotes, and non-ASCII characters.
+
 # 1.4.0, 2026-07-15
 
 ## Security

VERSION

@@ -1 +1 @@
-v1.4.0
+v1.5.0

benchmark/README.md

@@ -7,7 +7,14 @@ different components of ClickHouse Kafka connector. The test scope can be small
 
 ## How to Run
 
-As simple as:
+This is a standalone project that depends on the connector artifact by version,
+resolved from the local Maven repository. Publish the connector locally first
+(run from the repository root):
+```shell
+./gradlew publishToMavenLocal
+```
+
+Then, from this `benchmark` directory, run:
 ```shell
 ./gradlew run --args="-b kafka_connector.SimpleBenchmark -i 2 -t 10"
 ```
@@ -18,6 +25,33 @@ two (2) iterations for ten (10) seconds each.
 Two files will be created as output. One with `.json` extension is benchmark report that can be visualized with https://jmh.morethan.io/. 
 Another file with `.out` extension is standard output of the JMH. 
 
+### Insert benchmark (V1 vs V2)
+
+Use this benchmark to compare insert performance between client `V1` and `V2` for:
+- `rowbinary` inserts (schema records)
+- `json` inserts (schemaless map records)
+- `string` inserts (JSONEachRow payload)
+
+It runs with data sizes: `10000`, `50000`, `100000` rows.
+
+```shell
+./gradlew run --args="-b kafka_connector.InsertClientBenchmark -i 3 -t 15"
+```
+
+By default, the benchmark starts a temporary ClickHouse container via Testcontainers.
+You can point it to an external ClickHouse instance with environment variables:
+
+```shell
+CLICKHOUSE_HOST=localhost \
+CLICKHOUSE_PORT=8123 \
+CLICKHOUSE_USER=default \
+CLICKHOUSE_PASSWORD= \
+CLICKHOUSE_SSL=false \
+./gradlew run --args="-b kafka_connector.InsertClientBenchmark -i 3 -t 15"
+```
+
+Optional: override default Testcontainers image with `CLICKHOUSE_IMAGE`.
+
 ### Options 
 
 `-b <benchmark class>` - run specific benchmark. 

benchmark/build.gradle.kts

@@ -32,25 +32,20 @@ java {
 }
 
 dependencies {
+    // The connector is the single source of truth for the ClickHouse Java client, the Kafka
+    // Connect API, and (via its test fixtures) the Schema Registry / protobuf stack. They all
+    // arrive transitively as `api` dependencies; re-declaring any of them here would let the
+    // benchmark silently measure a different version set than the connector ships with.
     implementation("com.clickhouse.kafka:clickhouse-kafka-connect:${connectorVersion}")
-    implementation("com.clickhouse.kafka:clickhouse-kafka-connect:${connectorVersion}:test-fixtures")
-    implementation("org.apache.kafka:connect-api:2.7.0")
+    implementation(testFixtures("com.clickhouse.kafka:clickhouse-kafka-connect:${connectorVersion}"))
 
-    // https://mvnrepository.com/artifact/org.apache.commons/commons-lang3
-    implementation("org.apache.commons:commons-lang3:3.18.0")
+    // Used directly by the benchmarks themselves, versioned from the connector's catalog.
+    implementation(libs.testcontainers)
+    implementation(libs.testcontainers.clickhouse)
 
     implementation("commons-cli:commons-cli:1.5.0")
     implementation("org.openjdk.jmh:jmh-core:1.37")
     implementation("org.openjdk.jmh:jmh-generator-annprocess:1.37")
     annotationProcessor("org.openjdk.jmh:jmh-generator-annprocess:1.37")
-
-    implementation("com.google.protobuf:protobuf-java:3.25.1")
-    implementation("io.confluent:kafka-protobuf-serializer:7.9.1")
-    implementation("io.confluent:kafka-connect-protobuf-converter:7.9.1")
-
-//    // Schema Registry client for testing
-    implementation("io.confluent:kafka-schema-registry-client:7.5.4")
-    implementation("io.confluent:kafka-schema-registry:7.5.4")
-    implementation("io.confluent:kafka-schema-serializer:7.5.4")
 }
 

benchmark/gradle.properties

@@ -1,7 +1,7 @@
 # This file was generated by the Gradle 'init' task.
 # https://docs.gradle.org/current/userguide/build_environment.html#sec:gradle_configuration_properties
 
-org.gradle.configuration-cache=true
+org.gradle.configuration-cache=false
 org.gradle.parallel=true
 org.gradle.caching=true
 

benchmark/gradle/libs.versions.toml

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-9.2.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-9.3.0-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

benchmark/gradle/wrapper/gradle-wrapper.properties

@@ -1,9 +1,25 @@
 /*
- * This file was generated by the Gradle 'init' task.
- *
  * The settings file is used to specify which projects to include in your build.
  * For more detailed information on multi-project builds, please refer to https://docs.gradle.org/8.14.2/userguide/multi_project_builds.html in the Gradle documentation.
- * This project uses @Incubating APIs which are subject to change.
  */
 
 rootProject.name = "kafka-connector-benchmark"
+
+/*
+ * This is a standalone build. It does NOT use a Gradle composite build to consume
+ * the connector from the parent project. Instead, it depends on the published
+ * connector artifact by version, resolved from the local Maven repository (~/.m2).
+ *
+ * Before running the benchmark, publish the connector locally from the repo root:
+ *   ./gradlew publishToMavenLocal
+ */
+
+// Reuse the connector's version catalog so the few libraries the benchmark has to
+// declare itself stay pinned to the versions the connector was built against.
+dependencyResolutionManagement {
+    versionCatalogs {
+        create("libs") {
+            from(files("../gradle/libs.versions.toml"))
+        }
+    }
+}

benchmark/settings.gradle.kts

@@ -8,6 +8,8 @@
 import org.apache.commons.cli.OptionGroup;
 import org.apache.commons.cli.ParseException;
 import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.profile.GCProfiler;
+import org.openjdk.jmh.profile.MemPoolProfiler;
 import org.openjdk.jmh.results.format.ResultFormatType;
 import org.openjdk.jmh.runner.Runner;
 import org.openjdk.jmh.runner.RunnerException;
@@ -82,8 +84,8 @@ public static void main(String[] args) throws RunnerException {
                 .warmupTime(TimeValue.seconds(15))
                 .mode(Mode.SampleTime)
                 .timeUnit(TimeUnit.MILLISECONDS)
-//                .addProfiler(GCProfiler.class)
-//                .addProfiler(MemPoolProfiler.class)
+                .addProfiler(GCProfiler.class)
+                .addProfiler(MemPoolProfiler.class)
                 .jvmArgs("-Xms4g", "-Xmx4g")
                 .resultFormat(ResultFormatType.JSON)
                 .output(outputFile)