ClickHouseIO - Improve user agent (#37815)

* product name is deprecated, changing to client name

* create a share method + append external client name

* add unittests and rename original test file to IT

* reuse the same properties

* add user agent logic to the generated documentation

* fix query log targeted column

* add integration test to verify user agent is correct

* make sure integration tests are included

* move IT out of test task

Author: BentsiLeviav

.github/workflows/beam_PreCommit_Java_Clickhouse_IO_Direct.yml

@@ -92,6 +92,13 @@ jobs:
           arguments: |
             -PdisableSpotlessCheck=true \
             -PdisableCheckStyle=true \
+      - name: run Clickhouse IO IT script
+        uses: ./.github/actions/gradle-command-self-hosted-action
+        with:
+          gradle-command: :sdks:java:io:clickhouse:integrationTest
+          arguments: |
+            -PdisableSpotlessCheck=true \
+            -PdisableCheckStyle=true \
       - name: Archive JUnit Test Results
         uses: actions/upload-artifact@v4
         if: ${{ !success() }}

sdks/java/io/clickhouse/build.gradle

@@ -29,6 +29,8 @@ applyJavaNature(
   // javacc generated code produces lint warnings
   disableLintWarnings: ['dep-ann'],
 )
+provideIntegrationTestingDependencies()
+enableJavaPerformanceTesting()
 
 description = "Apache Beam :: SDKs :: Java :: IO :: ClickHouse"
 ext.summary = "IO to write to ClickHouse (https://clickhouse.com)."

sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIO.java

@@ -107,6 +107,18 @@
  * href="https://clickhouse.com/docs/guides/developer/deduplication">Deduplication strategies
  * documentation</a>
  *
+ * <h4>User agent</h4>
+ *
+ * <p>The connector automatically sets the ClickHouse client name to {@code Apache Beam/<version>},
+ * which is visible in {@code system.query_log.client_name}. If you set the {@code client_name}
+ * connection property, it is appended after the Beam identifier, for example:
+ *
+ * <pre>{@code
+ * Properties props = new Properties();
+ * props.setProperty("client_name", "MyApp/1.0");
+ * // Results in: "Apache Beam/<version> MyApp/1.0"
+ * }</pre>
+ *
  * <h4>Mapping between Beam and ClickHouse types</h4>
  *
  * <table summary="Type mapping">
@@ -244,21 +256,19 @@ public abstract static class Write<T> extends PTransform<PCollection<T>, PDone>
 
     @Override
     public PDone expand(PCollection<T> input) {
+
+      Properties properties = properties();
+      set(properties, "client_name", buildClientName(properties));
+
       TableSchema tableSchema = tableSchema();
       if (tableSchema == null) {
-        tableSchema = getTableSchema(clickHouseUrl(), database(), table(), properties());
+        tableSchema = getTableSchema(clickHouseUrl(), database(), table(), properties);
       }
 
-      String sdkVersion = ReleaseInfo.getReleaseInfo().getSdkVersion();
-      String userAgent = String.format("Apache Beam/%s", sdkVersion);
-
-      Properties properties = properties();
-
       set(properties, "max_insert_block_size", maxInsertBlockSize());
       set(properties, "insert_quorum", insertQuorum());
       set(properties, "insert_distributed_sync", insertDistributedSync());
       set(properties, "insert_deduplication", insertDeduplicate());
-      set(properties, "product_name", userAgent);
 
       WriteFn<T> fn =
           new AutoValue_ClickHouseIO_WriteFn.Builder<T>()
@@ -526,8 +536,7 @@ public void setup() throws Exception {
               .setPassword(password)
               .setDefaultDatabase(database())
               .setOptions(options)
-              .setClientName(
-                  String.format("Apache Beam/%s", ReleaseInfo.getReleaseInfo().getSdkVersion()));
+              .setClientName(properties().getProperty("client_name"));
 
       // Add optional compression if specified in properties
       String compress = properties().getProperty("compress", "false");
@@ -725,8 +734,7 @@ public static TableSchema getTableSchema(
               .setUsername(user)
               .setPassword(password)
               .setDefaultDatabase(database)
-              .setClientName(
-                  String.format("Apache Beam/%s", ReleaseInfo.getReleaseInfo().getSdkVersion()));
+              .setClientName(buildClientName(properties));
 
       try (Client client = clientBuilder.build()) {
         String query = "DESCRIBE TABLE " + quoteIdentifier(table);
@@ -766,6 +774,17 @@ public static TableSchema getTableSchema(
     }
   }
 
+  @VisibleForTesting
+  static String buildClientName(Properties properties) {
+    String beamAgent =
+        String.format("Apache Beam/%s", ReleaseInfo.getReleaseInfo().getSdkVersion());
+    String existingClientName = properties.getProperty("client_name");
+    if (!Strings.isNullOrEmpty(existingClientName)) {
+      return beamAgent + " " + existingClientName;
+    }
+    return beamAgent;
+  }
+
   static String quoteIdentifier(String identifier) {
     String backslash = "\\\\";
     String quote = "\"";