Error parsing Gtid when values exceeds MAX INTEGER

[subkanthi]

sink              | 2023-02-16 17:10:38,227 ERROR  ||  setAdditionalMetadata exception   [com.altinity.clickhouse.sink.connector.model.ClickHouseStruct]
sink              | java.lang.NumberFormatException: For input string: "9960950027"
sink              |     at java.base/java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
sink              |     at java.base/java.lang.Integer.parseInt(Integer.java:652)
sink              |     at java.base/java.lang.Integer.parseInt(Integer.java:770)
sink              |     at com.altinity.clickhouse.sink.connector.model.ClickHouseStruct.setAdditionalMetaData(ClickHouseStruct.java:207)
sink              |     at com.altinity.clickhouse.sink.connector.model.ClickHouseStruct.<init>(ClickHouseStruct.java:129)
sink              |     at com.altinity.clickhouse.sink.connector.converters.ClickHouseConverter.readBeforeOrAfterSection(ClickHouseConverter.java:274)
sink              |     at com.altinity.clickhouse.sink.connector.converters.ClickHouseConverter.convert(ClickHouseConverter.java:244)
sink              |     at com.altinity.clickhouse.sink.connector.ClickHouseSinkTask.put(ClickHouseSinkTask.java:107)
sink              |     at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:581)
sink              |     at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:333)
sink              |     at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:234)
sink              |     at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:203)
sink              |     at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:188)
sink              |     at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243)
sink              |     at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
sink              |     at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
sink              |     at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
sink              |     at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
sink              |     at java.base/java.lang.Thread.run(Thread.java:829)

[pservit]

my patch to use long instead of int for gtid

diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/metadata/TableMetaDataWriter.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/metadata/TableMetaDataWriter.java
index 21b0c38..44a7bcb 100644
--- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/metadata/TableMetaDataWriter.java
+++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/metadata/TableMetaDataWriter.java
@@ -56,7 +56,7 @@ public class TableMetaDataWriter {
         } else if (colName.equalsIgnoreCase(KafkaMetaData.SERVER_ID.getColumn())) {
             ps.setLong(index, record.getServerId()) ;
         } else if (colName.equalsIgnoreCase(KafkaMetaData.GTID.getColumn())) {
-            ps.setInt(index, record.getGtid());
+            ps.setLong(index, record.getGtid());
         } else if (colName.equalsIgnoreCase(KafkaMetaData.BINLOG_FILE.getColumn())) {
             ps.setString(index, record.getFile());
         } else if (colName.equalsIgnoreCase(KafkaMetaData.BINLOG_POSITION.getColumn())) {
diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/model/BlockMetaData.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/model/BlockMetaData.java
index bb19bef..4f646e1 100644
--- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/model/BlockMetaData.java
+++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/model/BlockMetaData.java
@@ -63,7 +63,7 @@ public class BlockMetaData {
 
     @Getter
     @Setter
-    int transactionId = -1;
+    long transactionId = -1;
 
     @Getter
     @Setter
@@ -90,7 +90,7 @@ public class BlockMetaData {
 
     public void update(ClickHouseStruct record) {
 
-        int gtId = record.getGtid();
+        long gtId = record.getGtid();
         if (gtId != -1) {
             if (gtId > this.transactionId) {
                 this.transactionId = gtId;
diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/model/ClickHouseStruct.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/model/ClickHouseStruct.java
index fe6af9b..0db522b 100644
--- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/model/ClickHouseStruct.java
+++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/model/ClickHouseStruct.java
@@ -76,7 +76,7 @@ public class ClickHouseStruct {
 
     @Getter
     @Setter
-    private int gtid = -1;
+    private long gtid = -1;
 
     @Getter
     @Setter
@@ -220,7 +220,7 @@ public class ClickHouseStruct {
             if(fieldNames.contains(GTID) && source.get(GTID) != null && source.get(GTID) instanceof String) {
                 String[] gtidArray = ((String) source.get(GTID)).split(":");
                 if(gtidArray.length == 2) {
-                    this.setGtid(Integer.parseInt(gtidArray[1]));
+                    this.setGtid(Long.parseLong(gtidArray[1]));
                 }
             }
             if(fieldNames.contains(LSN) && source.get(LSN) != null && source.get(LSN) instanceof Long) {

[subkanthi]

Fixed in #346