Update ClickHouseHelperClient.java (#238)

* Update ClickHouseHelperClient.java
ClickHouse · Nov 9, 2023 · b247a49 · b247a49
1 parent 8a2ef2a
commit b247a49
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 7 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,6 @@
+## 1.0.4 2023-11-08
+* Bugfix
+
 ## 1.0.3 2023-11-07
 * Added support for proxy configurations
 * Additional test cases

diff --git a/VERSION b/VERSION
@@ -1 +1 @@
-v1.0.3
+v1.0.4
diff --git a/src/main/java/com/clickhouse/kafka/connect/sink/db/helper/ClickHouseHelperClient.java b/src/main/java/com/clickhouse/kafka/connect/sink/db/helper/ClickHouseHelperClient.java
@@ -119,9 +119,11 @@ public ClickHouseResponse query(String query, ClickHouseFormat clickHouseFormat)
         int retryCount = 0;
         ClickHouseException ce = null;
         while (retryCount < retry) {
-            try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+            try (ClickHouseClient client = ClickHouseClient.builder()
+                    .options(getDefaultClientOptions())
+                    .nodeSelector(ClickHouseNodeSelector.of(ClickHouseProtocol.HTTP))
+                    .build();
                  ClickHouseResponse response = client.read(server)
-                         .options(getDefaultClientOptions())
                          .format(clickHouseFormat)
                          .query(query)
                          .executeAndWait()) {
@@ -138,9 +140,11 @@ public ClickHouseResponse query(String query, ClickHouseFormat clickHouseFormat)
 
     public List<String> showTables() {
         List<String> tablesNames = new ArrayList<>();
-        try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+        try (ClickHouseClient client = ClickHouseClient.builder()
+                .options(getDefaultClientOptions())
+                .nodeSelector(ClickHouseNodeSelector.of(ClickHouseProtocol.HTTP))
+                .build();
              ClickHouseResponse response = client.read(server)
-                     .options(getDefaultClientOptions())
                      .query("SHOW TABLES")
                      .executeAndWait()) {
             for (ClickHouseRecord r : response.records()) {
@@ -161,9 +165,11 @@ public Table describeTable(String tableName) {
         String describeQuery = String.format("DESCRIBE TABLE `%s`.`%s`", this.database, tableName);
         LOGGER.debug(describeQuery);
 
-        try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+        try (ClickHouseClient client = ClickHouseClient.builder()
+                .options(getDefaultClientOptions())
+                .nodeSelector(ClickHouseNodeSelector.of(ClickHouseProtocol.HTTP))
+                .build();
              ClickHouseResponse response = client.read(server)
-                     .options(getDefaultClientOptions())
                      .query(describeQuery)
                      .executeAndWait()) {
             Table table = new Table(tableName);