Update product name test to deel with cloud environment

Author: mzitnik

flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkTests.java

@@ -314,7 +314,8 @@ void ProductNameTest() throws Exception {
         if (ClickHouseServerForTests.isCloud())
             Thread.sleep(5000);
         // let's wait until data will be available in query log
-        String productName = ClickHouseServerForTests.extractProductName(ClickHouseServerForTests.getDatabase(), tableName);
+        String startWith = String.format("Flink-ClickHouse-Sink/%s", ClickHouseSinkVersion.getVersion());
+        String productName = ClickHouseServerForTests.extractProductName(ClickHouseServerForTests.getDatabase(), tableName, startWith);
         String compareString = String.format("Flink-ClickHouse-Sink/%s (fv:flink/%s, lv:scala/2.12)", ClickHouseSinkVersion.getVersion(), flinkVersion);
 
         boolean isContains = productName.contains(compareString);

flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/test/embedded/clickhouse/ClickHouseServerForTests.java

@@ -123,8 +123,8 @@ public static int countRows(String table) throws ExecutionException, Interrupted
         return countResult.get(0).getInteger(1);
     }
     // http_user_agent
-    public static String extractProductName(String databaseName, String tableName) {
-        String extractProductName = String.format("SELECT http_user_agent, tables FROM clusterAllReplicas('default', system.query_log) WHERE type = 'QueryStart' AND query_kind = 'Insert' AND has(databases,'%s') AND has(tables,'%s.%s') LIMIT 100", databaseName, databaseName, tableName);
+    public static String extractProductName(String databaseName, String tableName, String startWith) {
+        String extractProductName = String.format("SELECT http_user_agent, tables FROM clusterAllReplicas('default', system.query_log) WHERE type = 'QueryStart' AND query_kind = 'Insert' AND has(databases,'%s') AND has(tables,'%s.%s') and startsWith(http_user_agent, '%s') LIMIT 100", databaseName, databaseName, tableName, startWith);
         Client client = ClickHouseTestHelpers.getClient(host, port, isSSL, username, password);
         List<GenericRecord> userAgentResult = client.queryAll(extractProductName);
         if (!userAgentResult.isEmpty()) {

flink-connector-clickhouse-2.0.0/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkTests.java

@@ -308,7 +308,8 @@ void ProductNameTest() throws Exception {
         if (ClickHouseServerForTests.isCloud())
             Thread.sleep(5000);
         // let's wait until data will be available in query log
-        String productName = ClickHouseServerForTests.extractProductName(ClickHouseServerForTests.getDatabase(), tableName, "Flink-ClickHouse-Sink");
+        String startWith = String.format("Flink-ClickHouse-Sink/%s", ClickHouseSinkVersion.getVersion());
+        String productName = ClickHouseServerForTests.extractProductName(ClickHouseServerForTests.getDatabase(), tableName, startWith);
         String compareString = String.format("Flink-ClickHouse-Sink/%s (fv:flink/2.0.0, lv:scala/2.12)", ClickHouseSinkVersion.getVersion());
         boolean isContains = productName.contains(compareString);
         Assertions.assertTrue(isContains, "Expected user agent to contain: " + compareString + " but got: " + productName);

flink-connector-clickhouse-2.0.0/src/test/java/org/apache/flink/connector/test/embedded/clickhouse/ClickHouseServerForTests.java

@@ -124,15 +124,15 @@ public static int countRows(String table) throws ExecutionException, Interrupted
     }
 
     // http_user_agent
-    public static String extractProductName(String databaseName, String tableName, String productNameStartWith) {
-        String extractProductName = String.format("SELECT http_user_agent, tables FROM clusterAllReplicas('default', system.query_log) WHERE type = 'QueryStart' AND query_kind = 'Insert' AND has(databases,'%s') AND has(tables,'%s.%s') LIMIT 100", databaseName, databaseName, tableName);
+    public static String extractProductName(String databaseName, String tableName, String startWith) {
+        String extractProductName = String.format("SELECT http_user_agent, tables FROM clusterAllReplicas('default', system.query_log) WHERE type = 'QueryStart' AND query_kind = 'Insert' AND has(databases,'%s') AND has(tables,'%s.%s') and startsWith(http_user_agent, '%s') LIMIT 100", databaseName, databaseName, tableName, startWith);
         Client client = ClickHouseTestHelpers.getClient(host, port, isSSL, username, password);
         List<GenericRecord> userAgentResult = client.queryAll(extractProductName);
         String userAgentValue = null;
         if (!userAgentResult.isEmpty()) {
             for (GenericRecord userAgent : userAgentResult) {
                 userAgentValue = userAgent.getString(1);
-                if (userAgentValue.contains(productNameStartWith))
+                if (userAgentValue.contains(startWith))
                     return userAgent.getString(1);
             }
             throw new RuntimeException("Can not extract product name from " + userAgentValue);