add param for sqlserver e2e test

apache · Jun 15, 2023 · d7a87ff · d7a87ff
1 parent 8dc2a62
commit d7a87ff
Showing 1 changed file with 8 additions and 15 deletions.
diff --git a/...ver-cdc/src/main/java/com/ververica/cdc/connectors/sqlserver/source/offset/LsnOffset.java b/...ver-cdc/src/main/java/com/ververica/cdc/connectors/sqlserver/source/offset/LsnOffset.java
@@ -19,6 +19,7 @@
 import com.ververica.cdc.connectors.base.source.meta.offset.Offset;
 import io.debezium.connector.sqlserver.Lsn;
 import io.debezium.connector.sqlserver.SourceInfo;
+import io.debezium.connector.sqlserver.TxLogPosition;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -51,10 +52,6 @@ public LsnOffset(Lsn lsn) {
         this(lsn, null, null);
     }
 
-    public Lsn getLcn() {
-        return Lsn.valueOf(offset.get(SourceInfo.CHANGE_LSN_KEY));
-    }
-
     @Override
     public int compareTo(Offset offset) {
         LsnOffset that = (LsnOffset) offset;
@@ -69,17 +66,13 @@ public int compareTo(Offset offset) {
             return -1;
         }
 
-        Lsn lsn = this.getLcn();
-        Lsn targetLsn = that.getLcn();
-        if (targetLsn.isAvailable()) {
-            if (lsn.isAvailable()) {
-                return lsn.compareTo(targetLsn);
-            }
-            return -1;
-        } else if (lsn.isAvailable()) {
-            return 1;
-        }
-        return 0;
+        return TxLogPosition.valueOf(
+                        Lsn.valueOf(this.getOffset().get(SourceInfo.COMMIT_LSN_KEY)),
+                        Lsn.valueOf(this.getOffset().get(SourceInfo.CHANGE_LSN_KEY)))
+                .compareTo(
+                        TxLogPosition.valueOf(
+                                Lsn.valueOf(that.getOffset().get(SourceInfo.COMMIT_LSN_KEY)),
+                                Lsn.valueOf(that.getOffset().get(SourceInfo.CHANGE_LSN_KEY))));
     }
 
     @Override