Merge pull request #407 from askmyhat/main

Use getResultRows to check KeeperMap entry existance
ClickHouse · Jun 27, 2024 · d1fbe2b · d1fbe2b
2 parents 532e750 + b3345b3
commit d1fbe2b
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 2 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,8 @@
 * Bugfix to address string encoding issue
 * Bugfix to address issue with nested types and flatten_nested setting conflict
 * Bugfix to avoid storing keeper state in same column name if virtual topic is enabled
+* Updated java-client to 0.6.1
+* Bugfix to let create missing KeeperMap entries if there are some records present already
 * Added a flag to allow bypassing RowBinary and RowBinaryWithDefaults format for schema insertions
 * Bugfix to remove erroneous error messages about complex type handling
 

diff --git a/build.gradle.kts b/build.gradle.kts
@@ -51,7 +51,7 @@ repositories {
 
 extra.apply {
 
-    set("clickHouseDriverVersion", "0.6.0-patch4")
+    set("clickHouseDriverVersion", "0.6.1")
     set("kafkaVersion", "2.7.0")
     set("avroVersion", "1.9.2")
 

diff --git a/src/main/java/com/clickhouse/kafka/connect/sink/state/provider/KeeperStateProvider.java b/src/main/java/com/clickhouse/kafka/connect/sink/state/provider/KeeperStateProvider.java
@@ -85,7 +85,7 @@ public StateRecord getStateRecord(String topic, int partition) {
                      .query(selectStr)
                      .executeAndWait()) {
             LOGGER.debug("return size: {}", response.getSummary().getReadRows());
-            if ( response.getSummary().getReadRows() == 0) {
+            if ( response.getSummary().getResultRows() == 0) {
                 LOGGER.info(String.format("read state record: topic %s partition %s with NONE state", topic, partition));
                 return new StateRecord(topic, partition, 0, 0, State.NONE);
             }