Keep the sequence number state for cases where different batches coul…

…d have the same timestamp
Altinity · Apr 29, 2024 · 1ff600e · 1ff600e
1 parent 3f4b620
commit 1ff600e
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 7 deletions.
diff --git a/...c/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCapture.java b/...c/main/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCapture.java
@@ -10,6 +10,7 @@
 import com.altinity.clickhouse.sink.connector.ClickHouseSinkConnectorConfigVariables;
 import com.altinity.clickhouse.sink.connector.common.Metrics;
 import com.altinity.clickhouse.sink.connector.db.BaseDbWriter;
+import com.altinity.clickhouse.sink.connector.db.DBMetadata;
 import com.altinity.clickhouse.sink.connector.db.operations.ClickHouseAlterTable;
 import com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchExecutor;
 import com.altinity.clickhouse.sink.connector.executor.ClickHouseBatchRunnable;
@@ -177,7 +178,7 @@ private void performDDLOperation(String DDL, Properties props, SourceRecord sr,
 
         try {
             String clickHouseVersion = writer.getClickHouseVersion();
-            isNewReplacingMergeTreeEngine = new com.altinity.clickhouse.sink.connector.db.DBMetadata()
+            isNewReplacingMergeTreeEngine = new DBMetadata()
                     .checkIfNewReplacingMergeTree(clickHouseVersion);
         } catch (Exception e) {
             log.error("Error retrieving version");
@@ -752,6 +753,8 @@ private void appendToRecords(List<ClickHouseStruct> convertedRecords) {
 
     public static final long SEQUENCE_START = 1000000000;
     public static final long SEQUENCE_START_INITIAL = 500000000;
+
+    public static long sequenceNumber = SEQUENCE_START;
     /**
      * Function to add version to every record.
      * @param chStructs
@@ -764,12 +767,12 @@ public static void addVersion(List<ClickHouseStruct> chStructs, boolean initialS
             return;
         }
         long sequenceStartTime = chStructs.get(0).getTs_ms();
-        long sequence = SEQUENCE_START;
+        //long sequence = SEQUENCE_START;
         if(initialSeed) {
             // Add 500 million to the sequence
             // sequence += 500000000;
             // Add 000 to the debezium timestamp.
-            sequence = SEQUENCE_START_INITIAL;
+            sequenceNumber = SEQUENCE_START_INITIAL;
         }
         for(ClickHouseStruct chStruct: chStructs) {
             // Get the first ts_ms from chStruct
@@ -779,13 +782,13 @@ public static void addVersion(List<ClickHouseStruct> chStructs, boolean initialS
             // Get diff in seconds
             int diff = (int) (chStruct.getTs_ms() - sequenceStartTime) / 1000;
             if(diff > 1) {
-                sequence = SEQUENCE_START;
+                sequenceNumber = SEQUENCE_START;
                 sequenceStartTime = chStruct.getTs_ms();
             }   else {
-                sequence++;
+                sequenceNumber++;
             }
             // Pad the sequence number with 0s
-            chStruct.setSequenceNumber(chStruct.getTs_ms() * 100000 + sequence);
+            chStruct.setSequenceNumber(chStruct.getTs_ms() * 1000000 + sequenceNumber);
         }
     }
 }
diff --git a/...st/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCaptureTest.java b/...st/java/com/altinity/clickhouse/debezium/embedded/cdc/DebeziumChangeEventCaptureTest.java
@@ -111,7 +111,10 @@ public void shouldAssignUniqueSequenceNumbersWithinSameSecond() throws Interrupt
 
 
         // Validate ch5 and ch6
-        assertTrue(clickHouseStructs2.get(0).getSequenceNumber() != clickHouseStructs2.get(1).getSequenceNumber());
+        assertTrue(clickHouseStructs2.get(0).getSequenceNumber() < clickHouseStructs2.get(1).getSequenceNumber());
+
+        assertTrue(clickHouseStructs.get(3).getSequenceNumber() < clickHouseStructs2.get(0).getSequenceNumber());
+
 
     }