Merge pull request #407 from askmyhat/main

Use getResultRows to check KeeperMap entry existance

Author: Paultagoras

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
 

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")
 

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);
             }