DBZ-7905 Handle single position with host in gtid

debezium · May 31, 2024 · aeed110 · aeed110
1 parent 64e623f
commit aeed110
Show file tree

Hide file tree

Showing 3 changed files with 50 additions and 5 deletions.
diff --git a/src/main/java/io/debezium/connector/vitess/connection/VitessReplicationConnection.java b/src/main/java/io/debezium/connector/vitess/connection/VitessReplicationConnection.java
@@ -230,6 +230,7 @@ public void onNext(Vtgate.VStreamResponse response) {
             @Override
             public void onError(Throwable t) {
                 LOGGER.error("VStream streaming onError. Status: {}", Status.fromThrowable(t), t);
+                LOGGER.error("Error caused by", t.getCause());
                 // Only propagate the first error
                 error.compareAndSet(null, t);
                 reset();

diff --git a/src/main/java/io/debezium/connector/vitess/pipeline/txmetadata/Gtid.java b/src/main/java/io/debezium/connector/vitess/pipeline/txmetadata/Gtid.java
@@ -10,6 +10,8 @@
 import java.util.List;
 import java.util.Set;
 
+import io.debezium.DebeziumException;
+
 class Gtid {
 
     public String getVersion() {
@@ -52,18 +54,25 @@ private void initializeVersion(String transactionId) {
     }
 
     Gtid(String transactionId) {
-        initializeVersion(transactionId);
-        parseGtid(transactionId);
+        try {
+            initializeVersion(transactionId);
+            parseGtid(transactionId);
+        }
+        catch (Exception e) {
+            throw new DebeziumException("Error parsing GTID: " + transactionId, e);
+        }
     }
 
     private void parseGtid(String transactionId) {
         transactionId = trimVersion(transactionId);
         String[] transactions = transactionId.split(",");
         for (String transaction : transactions) {
-            String[] parts = transaction.split(":");
-            String hostname = parts[0];
+            String[] hostAndPositions = transaction.split(":");
+            String hostname = hostAndPositions[0];
             hosts.add(hostname);
-            String maxSequenceValue = parts[1].split("-")[1];
+            // This is either a range format eg 1-10 or a single position eg 8, either case we want the last number
+            String[] positions = hostAndPositions[1].split("-");
+            String maxSequenceValue = positions[positions.length - 1];
             sequenceValues.add(maxSequenceValue);
         }
     }

diff --git a/src/test/java/io/debezium/connector/vitess/pipeline/txmetadata/GtidTest.java b/src/test/java/io/debezium/connector/vitess/pipeline/txmetadata/GtidTest.java
@@ -6,12 +6,15 @@
 package io.debezium.connector.vitess.pipeline.txmetadata;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.util.List;
 import java.util.Set;
 
 import org.junit.Test;
 
+import io.debezium.DebeziumException;
+
 public class GtidTest {
 
     @Test
@@ -23,4 +26,36 @@ public void shouldInit() {
         assertThat(gtid.getHosts()).isEqualTo(Set.of("host1", "host2"));
     }
 
+    @Test
+    public void shouldHandleSingleValue() {
+        String expectedVersion = "MySQL56";
+        Gtid gtid = new Gtid(expectedVersion + "/host1:1,host2:2-10");
+        assertThat(gtid.getVersion()).isEqualTo(expectedVersion);
+        assertThat(gtid.getSequenceValues()).isEqualTo(List.of("1", "10"));
+        assertThat(gtid.getHosts()).isEqualTo(Set.of("host1", "host2"));
+    }
+
+    @Test
+    public void shouldThrowExceptionOnEmptyStringWithPrefix() {
+        String expectedVersion = "MySQL56";
+        assertThatThrownBy(() -> {
+            Gtid gtid = new Gtid(expectedVersion + "/");
+        }).isInstanceOf(DebeziumException.class);
+    }
+
+    @Test
+    public void shouldThrowExceptionOnVersionOnly() {
+        String expectedVersion = "MySQL56";
+        assertThatThrownBy(() -> {
+            Gtid gtid = new Gtid(expectedVersion);
+        }).isInstanceOf(DebeziumException.class);
+    }
+
+    @Test
+    public void shouldThrowExceptionOnVersionOnEmptyString() {
+        assertThatThrownBy(() -> {
+            Gtid gtid = new Gtid("");
+        }).isInstanceOf(DebeziumException.class);
+    }
+
 }