Support empty GTID set

Author: ilidemi

flow/connectors/mysql/replication_mechanism.go

@@ -18,10 +18,6 @@ type parsedReplicationOffset struct {
 }
 
 func parseReplicationOffsetText(flavor string, offsetText string) (parsedReplicationOffset, error) {
-	if offsetText == "" {
-		return parsedReplicationOffset{}, nil
-	}
-
 	if rest, isFilePos := strings.CutPrefix(offsetText, "!f:"); isFilePos {
 		comma := strings.LastIndexByte(rest, ',')
 		if comma == -1 {

flow/connectors/mysql/replication_mechanism_test.go

@@ -10,12 +10,6 @@ import (
 )
 
 func TestParseReplicationOffsetText(t *testing.T) {
-	t.Run("empty", func(t *testing.T) {
-		parsedOffset, err := parseReplicationOffsetText("mysql", "")
-		require.NoError(t, err)
-		require.Equal(t, parsedReplicationOffset{}, parsedOffset)
-	})
-
 	t.Run("filepos", func(t *testing.T) {
 		parsedOffset, err := parseReplicationOffsetText("mysql", "!f:mysql-bin.000001,4d2")
 		require.NoError(t, err)
@@ -42,6 +36,14 @@ func TestParseReplicationOffsetText(t *testing.T) {
 		require.Equal(t, mysql.Position{}, parsedOffset.pos)
 	})
 
+	t.Run("empty", func(t *testing.T) {
+		parsedOffset, err := parseReplicationOffsetText("mysql", "")
+		require.NoError(t, err)
+		require.Equal(t, protos.MySqlReplicationMechanism_MYSQL_GTID.String(), parsedOffset.mechanism)
+		require.True(t, parsedOffset.gset.IsEmpty())
+		require.Equal(t, mysql.Position{}, parsedOffset.pos)
+	})
+
 	t.Run("invalid gtid", func(t *testing.T) {
 		_, err := parseReplicationOffsetText("mysql", "not-a-valid-offset")
 		require.ErrorContains(t, err, `failed to parse mysql offset text "not-a-valid-offset" as GTID set`)