From 720920a3ce8e6ae59ebb8a33da62ac0cf0dbab67 Mon Sep 17 00:00:00 2001
From: okJiang <819421878@qq.com>
Date: Fri, 15 Mar 2024 19:23:11 +0800
Subject: [PATCH] DM/mariadb: sync the gtid executed in slave during
 master-slave replication (#10753)

close pingcap/tiflow#10741
---
 .../dm_mariadb_master_down_and_up.yaml        |  65 ++++++++
 dm/pkg/binlog/event/common.go                 |  18 ++-
 dm/pkg/binlog/event/common_test.go            |   2 +-
 dm/pkg/binlog/event/event.go                  |  37 +++--
 dm/pkg/binlog/event/event_test.go             |   9 +-
 dm/pkg/binlog/event/generator.go              |   6 +-
 dm/relay/local_reader.go                      |  16 +-
 dm/tests/mariadb_master_down_and_up/case.sh   | 150 ++++++++++++++++++
 .../conf/diff_config.toml                     |  34 ++++
 .../conf/dm-master.toml                       |   7 +
 .../conf/dm-worker1.toml                      |   2 +
 .../conf/source1.yaml                         |   9 ++
 .../conf/source1_relay.yaml                   |   9 ++
 .../conf/task-pessimistic.yaml                |  22 +++
 .../docker-compose.yml                        |  42 +++++
 dm/tests/mariadb_master_down_and_up/lib.sh    |  73 +++++++++
 go.mod                                        |   2 +-
 go.sum                                        |   4 +-
 18 files changed, 462 insertions(+), 45 deletions(-)
 create mode 100644 .github/workflows/dm_mariadb_master_down_and_up.yaml
 create mode 100644 dm/tests/mariadb_master_down_and_up/case.sh
 create mode 100644 dm/tests/mariadb_master_down_and_up/conf/diff_config.toml
 create mode 100644 dm/tests/mariadb_master_down_and_up/conf/dm-master.toml
 create mode 100644 dm/tests/mariadb_master_down_and_up/conf/dm-worker1.toml
 create mode 100644 dm/tests/mariadb_master_down_and_up/conf/source1.yaml
 create mode 100644 dm/tests/mariadb_master_down_and_up/conf/source1_relay.yaml
 create mode 100644 dm/tests/mariadb_master_down_and_up/conf/task-pessimistic.yaml
 create mode 100644 dm/tests/mariadb_master_down_and_up/docker-compose.yml
 create mode 100644 dm/tests/mariadb_master_down_and_up/lib.sh

diff --git a/.github/workflows/dm_mariadb_master_down_and_up.yaml b/.github/workflows/dm_mariadb_master_down_and_up.yaml
new file mode 100644
index 00000000000..1d431e25634
--- /dev/null
+++ b/.github/workflows/dm_mariadb_master_down_and_up.yaml
@@ -0,0 +1,65 @@
+name: Mariadb Master Down and Up
+
+on:
+  push:
+    branches:
+      - test-*
+  schedule:
+    - cron: '0 17-23 * * *' # run at minute 0 every hour from 01:00 ~ 07:00 UTC+8
+  workflow_dispatch:
+
+jobs:
+  mariadb-master-down-and-up:
+    name: mariadb-master-down-and-up
+    runs-on: ubuntu-20.04
+
+    steps:
+      - name: Set up Go env
+        uses: actions/setup-go@v3
+        with:
+          go-version: '1.21'
+
+      - name: Check out code
+        uses: actions/checkout@v2
+
+      - name: Cache go modules
+        uses: actions/cache@v2
+        with:
+          path: ~/go/pkg/mod
+          key: ${{ runner.os }}-ticdc-${{ hashFiles('go.sum') }}
+
+      - name: Cache Tools
+        id: cache-tools
+        uses: actions/cache@v2
+        with:
+          path: tools/bin
+          key: ${{ runner.os }}-ticdc-tools-${{ hashFiles('tools/check/go.sum') }}
+
+      - name: Build DM binary
+        run: make dm_integration_test_build
+
+      - name: Setup containers
+        run: |
+          docker-compose -f ./dm/tests/mariadb_master_down_and_up/docker-compose.yml up -d
+
+      - name: Run test cases
+        run: |
+          bash ./dm/tests/mariadb_master_down_and_up/case.sh
+
+      - name: Copy logs to hack permission
+        if: ${{ always() }}
+        run: |
+          mkdir ./logs
+          sudo cp -r -L /tmp/dm_test/mariadb_master_down_and_up/master/log ./logs/master
+          sudo cp -r -L /tmp/dm_test/mariadb_master_down_and_up/worker1/log ./logs/worker1
+          sudo chown -R runner ./logs
+
+      # Update logs as artifact seems not stable, so we set `continue-on-error: true` here.
+      - name: Upload logs
+        continue-on-error: true
+        uses: actions/upload-artifact@v2
+        if: ${{ always() }}
+        with:
+          name: upstream-switch-logs
+          path: |
+            ./logs
diff --git a/dm/pkg/binlog/event/common.go b/dm/pkg/binlog/event/common.go
index c5b6dfcb156..5d36153b1f1 100644
--- a/dm/pkg/binlog/event/common.go
+++ b/dm/pkg/binlog/event/common.go
@@ -163,7 +163,11 @@ func GTIDIncrease(flavor string, gSet gmysql.GTIDSet) (gmysql.GTIDSet, error) {
 		mariaGTID := singleGTID.(*gmysql.MariadbGTID)
 		mariaGTID.SequenceNumber++
 		gtidSet := new(gmysql.MariadbGTIDSet)
-		gtidSet.Sets = map[uint32]*gmysql.MariadbGTID{mariaGTID.DomainID: mariaGTID}
+		gtidSet.Sets = map[uint32]map[uint32]*gmysql.MariadbGTID{
+			mariaGTID.DomainID: {
+				mariaGTID.ServerID: mariaGTID,
+			},
+		}
 		clone = gtidSet
 	default:
 		err = terror.ErrBinlogGTIDSetNotValid.Generate(gSet, flavor)
@@ -203,11 +207,15 @@ func verifySingleGTID(flavor string, gSet gmysql.GTIDSet) (interface{}, error) {
 		if !ok {
 			return nil, terror.ErrBinlogGTIDMariaDBNotValid.Generate(gSet)
 		}
-		if len(mariaGTIDs.Sets) != 1 {
-			return nil, terror.ErrBinlogOnlyOneGTIDSupport.Generate(len(mariaGTIDs.Sets), gSet)
-		}
+		gtidCount := 0
 		var mariaGTID *gmysql.MariadbGTID
-		for _, mariaGTID = range mariaGTIDs.Sets {
+		for _, set := range mariaGTIDs.Sets {
+			gtidCount += len(set)
+			for _, mariaGTID = range set {
+			}
+		}
+		if gtidCount != 1 {
+			return nil, terror.ErrBinlogOnlyOneGTIDSupport.Generate(gtidCount, gSet)
 		}
 		return mariaGTID, nil
 	default:
diff --git a/dm/pkg/binlog/event/common_test.go b/dm/pkg/binlog/event/common_test.go
index 8213ef936e6..8098b6abfe2 100644
--- a/dm/pkg/binlog/event/common_test.go
+++ b/dm/pkg/binlog/event/common_test.go
@@ -71,7 +71,7 @@ func TestGenCommonFileHeader(t *testing.T) {
 
 	// MariaDB
 	flavor = gmysql.MariaDBFlavor
-	gSetStr = "1-2-12,2-2-3,3-3-8,4-4-4"
+	gSetStr = "1-2-12,2-2-3,3-3-8,3-4-9"
 
 	gSet, err = gtid.ParserGTID(flavor, gSetStr)
 	require.Nil(t, err)
diff --git a/dm/pkg/binlog/event/event.go b/dm/pkg/binlog/event/event.go
index 3273722a0a6..a0994de10c1 100644
--- a/dm/pkg/binlog/event/event.go
+++ b/dm/pkg/binlog/event/event.go
@@ -730,27 +730,32 @@ func GenMariaDBGTIDListEvent(header *replication.EventHeader, latestPos uint32,
 	payload := new(bytes.Buffer)
 
 	// Number of GTIDs, 4 bytes
-	numOfGTIDs := uint32(len(mariaDBGSet.Sets))
+	numOfGTIDs := uint32(0)
+	for _, set := range mariaDBGSet.Sets {
+		numOfGTIDs += uint32(len(set))
+	}
 	err := binary.Write(payload, binary.LittleEndian, numOfGTIDs)
 	if err != nil {
 		return nil, terror.ErrBinlogWriteBinaryData.AnnotateDelegate(err, "write Number of GTIDs %d", numOfGTIDs)
 	}
 
-	for _, mGTID := range mariaDBGSet.Sets {
-		// Replication Domain ID, 4 bytes
-		err = binary.Write(payload, binary.LittleEndian, mGTID.DomainID)
-		if err != nil {
-			return nil, terror.ErrBinlogWriteBinaryData.AnnotateDelegate(err, "write Replication Domain ID %d", mGTID.DomainID)
-		}
-		// Server_ID, 4 bytes
-		err = binary.Write(payload, binary.LittleEndian, mGTID.ServerID)
-		if err != nil {
-			return nil, terror.ErrBinlogWriteBinaryData.AnnotateDelegate(err, "write Server_ID %d", mGTID.ServerID)
-		}
-		// GTID sequence, 8 bytes
-		err = binary.Write(payload, binary.LittleEndian, mGTID.SequenceNumber)
-		if err != nil {
-			return nil, terror.ErrBinlogWriteBinaryData.AnnotateDelegate(err, "write GTID sequence %d", mGTID.SequenceNumber)
+	for _, set := range mariaDBGSet.Sets {
+		for _, mGTID := range set {
+			// Replication Domain ID, 4 bytes
+			err = binary.Write(payload, binary.LittleEndian, mGTID.DomainID)
+			if err != nil {
+				return nil, terror.ErrBinlogWriteBinaryData.AnnotateDelegate(err, "write Replication Domain ID %d", mGTID.DomainID)
+			}
+			// Server_ID, 4 bytes
+			err = binary.Write(payload, binary.LittleEndian, mGTID.ServerID)
+			if err != nil {
+				return nil, terror.ErrBinlogWriteBinaryData.AnnotateDelegate(err, "write Server_ID %d", mGTID.ServerID)
+			}
+			// GTID sequence, 8 bytes
+			err = binary.Write(payload, binary.LittleEndian, mGTID.SequenceNumber)
+			if err != nil {
+				return nil, terror.ErrBinlogWriteBinaryData.AnnotateDelegate(err, "write GTID sequence %d", mGTID.SequenceNumber)
+			}
 		}
 	}
 
diff --git a/dm/pkg/binlog/event/event_test.go b/dm/pkg/binlog/event/event_test.go
index 7fa80e5af14..87dbc835960 100644
--- a/dm/pkg/binlog/event/event_test.go
+++ b/dm/pkg/binlog/event/event_test.go
@@ -519,7 +519,6 @@ func TestGenRowsEvent(t *testing.T) {
 	require.Equal(t, tableID, rowsEvBody.TableID)
 	require.Equal(t, uint64(len(rows[0])), rowsEvBody.ColumnCount)
 	require.Equal(t, 0, rowsEvBody.Version) // WRITE_ROWS_EVENTv0
-	require.Nil(t, rowsEvBody.ExtraData)
 	require.Equal(t, rows, rowsEvBody.Rows)
 
 	// multi rows, with different length, invalid
@@ -664,10 +663,10 @@ func TestGenMariaDBGTIDListEvent(t *testing.T) {
 	require.True(t, ok)
 	require.NotNil(t, gtidListEvBody)
 	require.Len(t, gtidListEvBody.GTIDs, 1)
-	require.Equal(t, *mGSet.Sets[gtidListEvBody.GTIDs[0].DomainID], gtidListEvBody.GTIDs[0])
+	require.Equal(t, *mGSet.Sets[gtidListEvBody.GTIDs[0].DomainID][gtidListEvBody.GTIDs[0].ServerID], gtidListEvBody.GTIDs[0])
 
 	// valid gSet with multi GTIDs
-	gSet, err = gtid.ParserGTID(gmysql.MariaDBFlavor, "1-2-12,2-2-3,3-3-8,4-4-4")
+	gSet, err = gtid.ParserGTID(gmysql.MariaDBFlavor, "1-2-12,2-2-3,3-3-8,3-4-4")
 	require.Nil(t, err)
 	require.NotNil(t, gSet)
 	mGSet, ok = gSet.(*gmysql.MariadbGTIDSet)
@@ -684,7 +683,9 @@ func TestGenMariaDBGTIDListEvent(t *testing.T) {
 	require.NotNil(t, gtidListEvBody)
 	require.Len(t, gtidListEvBody.GTIDs, 4)
 	for _, mGTID := range gtidListEvBody.GTIDs {
-		mGTID2, ok := mGSet.Sets[mGTID.DomainID]
+		set, ok := mGSet.Sets[mGTID.DomainID]
+		require.True(t, ok)
+		mGTID2, ok := set[mGTID.ServerID]
 		require.True(t, ok)
 		require.Equal(t, *mGTID2, mGTID)
 	}
diff --git a/dm/pkg/binlog/event/generator.go b/dm/pkg/binlog/event/generator.go
index a2a4bb2b95c..ac7dca4605c 100644
--- a/dm/pkg/binlog/event/generator.go
+++ b/dm/pkg/binlog/event/generator.go
@@ -86,7 +86,11 @@ func newGenerator(flavor, version string, serverID uint32, latestPos uint32, lat
 		if !ok || prevGSet == nil {
 			return nil, terror.ErrBinlogGTIDMariaDBNotValid.Generate(previousGTIDs)
 		}
-		prevGTID, ok := prevGSet.Sets[mariaGTID.DomainID]
+		set, ok := prevGSet.Sets[mariaGTID.DomainID]
+		if !ok {
+			return nil, terror.ErrBinlogLatestGTIDNotInPrev.Generate(latestGTID, previousGTIDs)
+		}
+		prevGTID, ok := set[mariaGTID.ServerID]
 		if !ok || prevGTID.ServerID != mariaGTID.ServerID || prevGTID.SequenceNumber != mariaGTID.SequenceNumber {
 			return nil, terror.ErrBinlogLatestGTIDNotInPrev.Generate(latestGTID, previousGTIDs)
 		}
diff --git a/dm/relay/local_reader.go b/dm/relay/local_reader.go
index 2c8524fb244..2f23dd3c345 100644
--- a/dm/relay/local_reader.go
+++ b/dm/relay/local_reader.go
@@ -548,21 +548,7 @@ func (r *BinlogReader) parseFile(
 				Pos:  uint32(ev.Position),
 			}
 			r.tctx.L().Info("rotate binlog", zap.Stringer("position", currentPos))
-		case *replication.GTIDEvent:
-			if r.prevGset == nil {
-				state.latestPos = int64(e.Header.LogPos)
-				break
-			}
-			gtidStr, err2 := event.GetGTIDStr(e)
-			if err2 != nil {
-				return errors.Trace(err2)
-			}
-			state.skipGTID, err = r.advanceCurrentGtidSet(gtidStr)
-			if err != nil {
-				return errors.Trace(err)
-			}
-			state.latestPos = int64(e.Header.LogPos)
-		case *replication.MariadbGTIDEvent:
+		case *replication.GTIDEvent, *replication.MariadbGTIDEvent:
 			if r.prevGset == nil {
 				state.latestPos = int64(e.Header.LogPos)
 				break
diff --git a/dm/tests/mariadb_master_down_and_up/case.sh b/dm/tests/mariadb_master_down_and_up/case.sh
new file mode 100644
index 00000000000..e678d3bcfd2
--- /dev/null
+++ b/dm/tests/mariadb_master_down_and_up/case.sh
@@ -0,0 +1,150 @@
+#!/bin/bash
+
+set -exu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+PATH=$CUR/../_utils:$PATH # for sync_diff_inspector
+
+source $CUR/lib.sh
+
+function clean_data() {
+	echo "-------clean_data--------"
+
+	exec_sql $slave_port "stop slave;"
+	exec_sql $slave_port "reset master;"
+
+	exec_sql $master_port "drop database if exists db1;"
+	exec_sql $master_port "drop database if exists db2;"
+	exec_sql $master_port "drop database if exists ${db};"
+	exec_sql $slave_port "drop database if exists db1;"
+	exec_sql $slave_port "drop database if exists db2;"
+	exec_sql $slave_port "drop database if exists ${db};"
+	exec_sql $slave_port "reset master;"
+	exec_tidb $tidb_port "drop database if exists db1;"
+	exec_tidb $tidb_port "drop database if exists db2;"
+	exec_tidb $tidb_port "drop database if exists ${db};"
+	rm -rf /tmp/dm_test
+}
+
+function cleanup_process() {
+	echo "-------cleanup_process--------"
+	pkill -hup dm-worker.test 2>/dev/null || true
+	pkill -hup dm-master.test 2>/dev/null || true
+	pkill -hup dm-syncer.test 2>/dev/null || true
+}
+
+function setup_replica() {
+	echo "-------setup_replica--------"
+
+	master_status=($(get_master_status))
+	master_gtid=$(exec_sql $master_port "select binlog_gtid_pos('${master_status[0]}', ${master_status[1]})" | awk 'NR==2')
+	exec_sql $slave_port "set global gtid_slave_pos = '$master_gtid';"
+
+	# master --> slave
+	change_master_to_gtid $slave_port $master_port
+}
+
+function run_dm_components_and_create_source() {
+	echo "-------run_dm_components--------"
+
+	pkill -9 dm-master || true
+	pkill -9 dm-worker || true
+
+	run_dm_master $WORK_DIR/master $MASTER_PORT $CUR/conf/dm-master.toml
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"list-member" \
+		"alive" 1
+
+	run_dm_worker $WORK_DIR/worker1 $WORKER1_PORT $CUR/conf/dm-worker1.toml
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"list-member" \
+		"free" 1
+	if [ "$1" = "relay" ]; then
+		run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+			"operate-source create $CUR/conf/source1_relay.yaml" \
+			"\"result\": true" 2
+	else
+		run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+			"operate-source create $CUR/conf/source1.yaml" \
+			"\"result\": true" 2
+	fi
+
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"list-member" \
+		"alive" 1 \
+		"bound" 1
+}
+
+function gen_full_data() {
+	echo "-------gen_full_data--------"
+
+	exec_sql $master_port "create database ${db} collate latin1_bin;"
+	exec_sql $master_port "create table ${db}.${tb}(id int primary key, a int);"
+	for i in $(seq 1 100); do
+		exec_sql $master_port "insert into ${db}.${tb} values($i,$i);"
+	done
+}
+
+function start_task() {
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"start-task $CUR/conf/task-pessimistic.yaml --remove-meta" \
+		"\"result\": true" 2
+}
+
+function verify_result() {
+	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+}
+
+function clean_task() {
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"stop-task task_pessimistic" \
+		"\"result\": true" 2
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"operate-source stop mysql-replica-01" \
+		"\"result\": true" 2
+}
+
+function test_master_down_and_up() {
+	cleanup_process
+	clean_data
+	install_sync_diff
+	setup_replica
+	gen_full_data
+	run_dm_components_and_create_source $1
+	start_task
+	verify_result
+	echo "-------start test--------"
+
+	for i in $(seq 201 250); do
+		exec_sql $master_port "insert into ${db}.${tb} values($i,$i);"
+	done
+	verify_result
+
+	# make master down
+	docker-compose -f $CUR/docker-compose.yml pause mariadb_master
+	# execute sqls in slave
+	for i in $(seq 401 450); do
+		exec_sql $slave_port "insert into ${db}.${tb} values($i,$i);"
+	done
+	verify_result
+
+	# make master up
+	docker-compose -f $CUR/docker-compose.yml unpause mariadb_master
+	for i in $(seq 501 550); do
+		exec_sql $master_port "insert into ${db}.${tb} values($i,$i);"
+	done
+
+	verify_result
+
+	clean_task
+	echo "CASE=test_master_down_and_up $1 success"
+}
+
+function run() {
+	wait_mysql 3306 1
+	wait_mysql 3307 2
+	test_master_down_and_up no_relay
+	test_master_down_and_up relay
+}
+
+run
diff --git a/dm/tests/mariadb_master_down_and_up/conf/diff_config.toml b/dm/tests/mariadb_master_down_and_up/conf/diff_config.toml
new file mode 100644
index 00000000000..21ea1231e60
--- /dev/null
+++ b/dm/tests/mariadb_master_down_and_up/conf/diff_config.toml
@@ -0,0 +1,34 @@
+check-thread-count = 4
+
+export-fix-sql = true
+
+check-struct-only = false
+
+[task]
+    output-dir = "/tmp/ticdc_dm_test/output"
+
+    source-instances = ["mysql1"]
+
+    target-instance = "tidb0"
+
+    target-check-tables = ["db_pessimistic.tb"]
+
+[routes.rule1]
+schema-pattern = "db_pessimistic"
+table-pattern = "tb*"
+target-schema = "db_pessimistic"
+target-table = "tb"
+
+[data-sources]
+[data-sources.mysql1]
+host = "127.0.0.1"
+port = 3307
+user = "root"
+password = "123456"
+route-rules = ["rule1"]
+
+[data-sources.tidb0]
+host = "127.0.0.1"
+port = 4000
+user = "root"
+password = ""
diff --git a/dm/tests/mariadb_master_down_and_up/conf/dm-master.toml b/dm/tests/mariadb_master_down_and_up/conf/dm-master.toml
new file mode 100644
index 00000000000..e700eeffa5c
--- /dev/null
+++ b/dm/tests/mariadb_master_down_and_up/conf/dm-master.toml
@@ -0,0 +1,7 @@
+# Master Configuration.
+name = "master1"
+master-addr = ":8261"
+advertise-addr = "127.0.0.1:8261"
+peer-urls = "127.0.0.1:8291"
+initial-cluster = "master1=http://127.0.0.1:8291"
+auto-compaction-retention = "3s"
diff --git a/dm/tests/mariadb_master_down_and_up/conf/dm-worker1.toml b/dm/tests/mariadb_master_down_and_up/conf/dm-worker1.toml
new file mode 100644
index 00000000000..7a72ea72bf8
--- /dev/null
+++ b/dm/tests/mariadb_master_down_and_up/conf/dm-worker1.toml
@@ -0,0 +1,2 @@
+name = "worker1"
+join = "127.0.0.1:8261"
diff --git a/dm/tests/mariadb_master_down_and_up/conf/source1.yaml b/dm/tests/mariadb_master_down_and_up/conf/source1.yaml
new file mode 100644
index 00000000000..202139cfea3
--- /dev/null
+++ b/dm/tests/mariadb_master_down_and_up/conf/source1.yaml
@@ -0,0 +1,9 @@
+source-id: "mysql-replica-01"
+enable-gtid: true
+enable-relay: false
+
+from:
+  host: "127.0.0.1"
+  user: "root"
+  password: "123456"
+  port: 3307
diff --git a/dm/tests/mariadb_master_down_and_up/conf/source1_relay.yaml b/dm/tests/mariadb_master_down_and_up/conf/source1_relay.yaml
new file mode 100644
index 00000000000..d9e5eaf6852
--- /dev/null
+++ b/dm/tests/mariadb_master_down_and_up/conf/source1_relay.yaml
@@ -0,0 +1,9 @@
+source-id: "mysql-replica-01"
+enable-gtid: true
+enable-relay: true
+
+from:
+  host: "127.0.0.1"
+  user: "root"
+  password: "123456"
+  port: 3307
diff --git a/dm/tests/mariadb_master_down_and_up/conf/task-pessimistic.yaml b/dm/tests/mariadb_master_down_and_up/conf/task-pessimistic.yaml
new file mode 100644
index 00000000000..3a12a864e2c
--- /dev/null
+++ b/dm/tests/mariadb_master_down_and_up/conf/task-pessimistic.yaml
@@ -0,0 +1,22 @@
+---
+name: "task_pessimistic"
+task-mode: all
+shard-mode: pessimistic
+
+target-database:
+  host: "127.0.0.1"
+  port: 4000
+  user: "root"
+  password: ""
+
+mysql-instances:
+  -
+    source-id: "mysql-replica-01"
+    black-white-list:  "instance"
+    mydumper-thread: 4
+    loader-thread: 16
+    syncer-thread: 16
+
+black-white-list:
+  instance:
+    do-dbs: ["db_pessimistic"]
diff --git a/dm/tests/mariadb_master_down_and_up/docker-compose.yml b/dm/tests/mariadb_master_down_and_up/docker-compose.yml
new file mode 100644
index 00000000000..e60a02156e4
--- /dev/null
+++ b/dm/tests/mariadb_master_down_and_up/docker-compose.yml
@@ -0,0 +1,42 @@
+version: "3.7"
+
+networks:
+  db-networks:
+    ipam: # for static IP
+      config:
+        - subnet: "172.28.128.0/24"
+
+# re-usable extension fields, https://docs.docker.com/compose/compose-file/#extension-fields.
+x-node: &default-node
+  privileged: true
+  restart: always
+  environment:
+    MYSQL_ROOT_PASSWORD: "123456"
+
+# In this case, gtid_domain_id should be the same, and gtid_strict_mode should be OFF.
+services:
+  mariadb_master:
+    <<: *default-node
+    image: mariadb:10.1
+    command: --log-bin=/var/lib/mysql/mysql-bin --log_slave_updates=ON --server-id=1 --gtid_domain_id=0 --binlog-format=ROW
+    container_name: mariadb_master
+    hostname: mariadb_master
+    ports:
+      - "3306:3306"
+
+  mariadb_slave:
+    <<: *default-node
+    image: mariadb:10.1
+    command: --log-bin=/var/lib/mysql/mysql-bin --log_slave_updates=ON --server-id=2 --gtid_domain_id=0 --binlog-format=ROW
+    container_name: mariadb_slave
+    hostname: mariadb_slave
+    ports:
+      - "3307:3306"
+
+  tidb: # downstream TiDB
+    container_name: tidb
+    hostname: tidb
+    ports:
+      - "4000:4000"
+    image: pingcap/tidb:latest
+    restart: always
diff --git a/dm/tests/mariadb_master_down_and_up/lib.sh b/dm/tests/mariadb_master_down_and_up/lib.sh
new file mode 100644
index 00000000000..3d38de273e7
--- /dev/null
+++ b/dm/tests/mariadb_master_down_and_up/lib.sh
@@ -0,0 +1,73 @@
+#!/bin/bash
+
+set -eu
+
+export TEST_DIR=/tmp/dm_test
+export TEST_NAME="mariadb_master_down_and_up"
+export DM_MASTER_EXTRA_ARG=""
+
+WORK_DIR=$TEST_DIR/$TEST_NAME
+rm -rf $WORK_DIR
+mkdir -p $WORK_DIR
+
+db="db_pessimistic"
+tb="tb"
+
+master_port="3306"
+slave_port="3307"
+tidb_port="4000"
+MASTER_PORT=8261
+WORKER1_PORT=8262
+
+function exec_sql() {
+	echo $2 | MYSQL_PWD=123456 mysql -uroot -h127.0.0.1 -P$1
+}
+
+function exec_tidb() {
+	echo $2 | mysql -uroot -h127.0.0.1 -P$1
+}
+
+function install_sync_diff() {
+	curl https://download.pingcap.org/tidb-enterprise-tools-nightly-linux-amd64.tar.gz | tar xz
+	mkdir -p bin
+	mv tidb-enterprise-tools-nightly-linux-amd64/bin/sync_diff_inspector bin/
+}
+
+function get_master_status() {
+	arr=$(echo "show master status;" | MYSQL_PWD=123456 mysql -uroot -h127.0.0.1 -P3306 | awk 'NR==2')
+	echo $arr
+}
+
+function change_master_to_gtid() {
+	exec_sql $1 "stop slave;"
+	exec_sql $1 "change master to master_host='mariadb_master',master_port=$2,master_user='root',master_password='123456',master_use_gtid=slave_pos;"
+	exec_sql $1 "start slave;"
+}
+
+function wait_mysql() {
+	echo "-------wait_mysql--------"
+
+	i=0
+	while ! mysqladmin -h127.0.0.1 -P$1 -uroot ping --connect-timeout=1 >/dev/null 2>&1; do
+		echo "wait mysql"
+		i=$((i + 1))
+		if [ "$i" -gt 20 ]; then
+			echo "wait for mysql $1:3306 timeout"
+			exit 1
+		fi
+		sleep 1
+	done
+	i=0
+
+	server_id=$(echo "show variables like 'server_id';" | MYSQL_PWD=123456 mysql -uroot -h127.0.0.1 -P$1 | awk 'NR==2' | awk '{print $2}')
+	while [ "$server_id" != $2 ]; do
+		echo "wait server_id"
+		i=$((i + 1))
+		if [ "$i" -gt 20 ]; then
+			echo "different server_id: $server_id, expect: $2, host: $1"
+			exit 1
+		fi
+		sleep 1
+		server_id=$(echo "show variables like 'server_id';" | MYSQL_PWD=123456 mysql -uroot -h127.0.0.1 -P$1 | awk 'NR==2' | awk '{print $2}')
+	done
+}
diff --git a/go.mod b/go.mod
index 8419980d9ea..fed65c0584d 100644
--- a/go.mod
+++ b/go.mod
@@ -33,7 +33,7 @@ require (
 	github.com/gin-gonic/gin v1.9.1
 	github.com/glebarez/go-sqlite v1.21.2
 	github.com/glebarez/sqlite v1.7.0
-	github.com/go-mysql-org/go-mysql v1.7.1-0.20230619063055-fd67d94318fd
+	github.com/go-mysql-org/go-mysql v1.7.1-0.20240314115043-2199dfb0ba98
 	github.com/go-oauth2/oauth2/v4 v4.5.2
 	github.com/go-ozzo/ozzo-validation/v4 v4.3.0
 	github.com/go-sql-driver/mysql v1.7.1
diff --git a/go.sum b/go.sum
index ba4834c2f0b..39d989d16f4 100644
--- a/go.sum
+++ b/go.sum
@@ -366,8 +366,8 @@ github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ4
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-martini/martini v0.0.0-20170121215854-22fa46961aab/go.mod h1:/P9AEU963A2AYjv4d1V5eVL1CQbEJq6aCNHDDjibzu8=
-github.com/go-mysql-org/go-mysql v1.7.1-0.20230619063055-fd67d94318fd h1:lqWdv8GEYqF1deivEmnSx81GfcAUZ/FoxilGxm/kwWs=
-github.com/go-mysql-org/go-mysql v1.7.1-0.20230619063055-fd67d94318fd/go.mod h1:kOk/pFv3q5EPspyQfDRGLmEA6wfMvIeV4DmThwzkNzs=
+github.com/go-mysql-org/go-mysql v1.7.1-0.20240314115043-2199dfb0ba98 h1:ygai+7MZSkAPhRwRraWf1EsPTgrz4U2+x3NVWzJYQys=
+github.com/go-mysql-org/go-mysql v1.7.1-0.20240314115043-2199dfb0ba98/go.mod h1:kwbF156Z9Sy8amP3E1SZp7/s/0PuJj/xKaOWToQiq0Y=
 github.com/go-oauth2/oauth2/v4 v4.5.2 h1:CuZhD3lhGuI6aNLyUbRHXsgG2RwGRBOuCBfd4WQKqBQ=
 github.com/go-oauth2/oauth2/v4 v4.5.2/go.mod h1:wk/2uLImWIa9VVQDgxz99H2GDbhmfi/9/Xr+GvkSUSQ=
 github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=