syncer: update dependency to avoid exit when meet unsupported binlog (#…

…1707) (#1734)
pingcap · Jun 8, 2021 · 11bdda2 · 11bdda2
1 parent 3963412
commit 11bdda2
Show file tree

Hide file tree

Showing 8 changed files with 34 additions and 5 deletions.
diff --git a/_utils/terror_gen/errors_release.txt b/_utils/terror_gen/errors_release.txt
@@ -319,6 +319,7 @@ ErrSyncerOperatorNotExist,[code=36065:class=sync-unit:scope=internal:level=low],
 ErrSyncerReplaceEventNotExist,[code=36066:class=sync-unit:scope=internal:level=high], "Message: replace event not exist, location: %s"
 ErrSyncerParseDDL,[code=36067:class=sync-unit:scope=internal:level=high], "Message: parse DDL: %s, Workaround: Please confirm your DDL statement is correct and needed. For TiDB compatible DDL, see https://docs.pingcap.com/tidb/stable/mysql-compatibility#ddl. You can use `handle-error` command to skip or replace the DDL or add a binlog filter rule to ignore it if the DDL is not needed."
 ErrSyncerUnsupportedStmt,[code=36068:class=sync-unit:scope=internal:level=high], "Message: `%s` statement not supported in %s mode"
+ErrSyncerGetEvent,[code=36069:class=sync-unit:scope=upstream:level=high], "Message: get binlog event error: %v, Workaround: Please check if the binlog file could be parsed by `mysqlbinlog`."
 ErrMasterSQLOpNilRequest,[code=38001:class=dm-master:scope=internal:level=medium], "Message: nil request not valid"
 ErrMasterSQLOpNotSupport,[code=38002:class=dm-master:scope=internal:level=medium], "Message: op %s not supported"
 ErrMasterSQLOpWithoutSharding,[code=38003:class=dm-master:scope=internal:level=medium], "Message: operate request without --sharding specified not valid"

diff --git a/errors.toml b/errors.toml
@@ -1924,6 +1924,12 @@ description = ""
 workaround = ""
 tags = ["internal", "high"]
 
+[error.DM-sync-unit-36069]
+message = "get binlog event error: %v"
+description = ""
+workaround = "Please check if the binlog file could be parsed by `mysqlbinlog`."
+tags = ["upstream", "high"]
+
 [error.DM-dm-master-38001]
 message = "nil request not valid"
 description = ""

diff --git a/go.mod b/go.mod
@@ -8,7 +8,7 @@ require (
 	github.com/coreos/go-semver v0.3.0
 	github.com/docker/go-units v0.4.0
 	github.com/dustin/go-humanize v1.0.0
-	github.com/go-mysql-org/go-mysql v1.1.3-0.20210420093540-a51d9dec4f96
+	github.com/go-mysql-org/go-mysql v1.1.3-0.20210519094347-5ec8d511f5be
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/gogo/gateway v1.1.0
 	github.com/gogo/protobuf v1.3.2

diff --git a/go.sum b/go.sum
@@ -303,8 +303,8 @@ github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
-github.com/go-mysql-org/go-mysql v1.1.3-0.20210420093540-a51d9dec4f96 h1:+0HKUBVZNwqh+k20hbdiFoZQ7neXijUm54G/B2cQVgM=
-github.com/go-mysql-org/go-mysql v1.1.3-0.20210420093540-a51d9dec4f96/go.mod h1:3lFZKf7l95Qo70+3XB2WpiSf9wu2s3na3geLMaIIrqQ=
+github.com/go-mysql-org/go-mysql v1.1.3-0.20210519094347-5ec8d511f5be h1:12z4p/lldNvXStdf2qPICQ3eyTC0B0kw32X8ysx40Yk=
+github.com/go-mysql-org/go-mysql v1.1.3-0.20210519094347-5ec8d511f5be/go.mod h1:3lFZKf7l95Qo70+3XB2WpiSf9wu2s3na3geLMaIIrqQ=
 github.com/go-ole/go-ole v1.2.4 h1:nNBDSCOigTSiarFpYE9J/KtEA1IOW4CNeqT9TQDqCxI=
 github.com/go-ole/go-ole v1.2.4/go.mod h1:XCwSNxSkXRo4vlyPy93sltvi/qJq0jqQhjqQNIwKuxM=
 github.com/go-openapi/jsonpointer v0.17.0/go.mod h1:cOnomiV+CVVwFLk0A/MExoFMjwdsUdVpsRhURCKh+3M=

diff --git a/pkg/terror/error_list.go b/pkg/terror/error_list.go
@@ -412,6 +412,7 @@ const (
 	codeSyncerReplaceEventNotExist
 	codeSyncerParseDDL
 	codeSyncerUnsupportedStmt
+	codeSyncerGetEvent
 )
 
 // DM-master error code.
@@ -1005,6 +1006,7 @@ var (
 	ErrSyncerReplaceEventNotExist           = New(codeSyncerReplaceEventNotExist, ClassSyncUnit, ScopeInternal, LevelHigh, "replace event not exist, location: %s", "")
 	ErrSyncerParseDDL                       = New(codeSyncerParseDDL, ClassSyncUnit, ScopeInternal, LevelHigh, "parse DDL: %s", "Please confirm your DDL statement is correct and needed. For TiDB compatible DDL, see https://docs.pingcap.com/tidb/stable/mysql-compatibility#ddl. You can use `handle-error` command to skip or replace the DDL or add a binlog filter rule to ignore it if the DDL is not needed.")
 	ErrSyncerUnsupportedStmt                = New(codeSyncerUnsupportedStmt, ClassSyncUnit, ScopeInternal, LevelHigh, "`%s` statement not supported in %s mode", "")
+	ErrSyncerGetEvent                       = New(codeSyncerGetEvent, ClassSyncUnit, ScopeUpstream, LevelHigh, "get binlog event error: %v", "Please check if the binlog file could be parsed by `mysqlbinlog`.")
 
 	// DM-master error.
 	ErrMasterSQLOpNilRequest        = New(codeMasterSQLOpNilRequest, ClassDMMaster, ScopeInternal, LevelMedium, "nil request not valid", "")

diff --git a/syncer/streamer_controller.go b/syncer/streamer_controller.go
@@ -15,6 +15,7 @@ package syncer
 
 import (
 	"context"
+	"errors"
 	"strings"
 	"sync"
 	"time"
@@ -249,6 +250,9 @@ func (c *StreamerController) GetEvent(tctx *tcontext.Context) (event *replicatio
 
 	event, err = streamer.GetEvent(ctx)
 	cancel()
+	failpoint.Inject("GetEventError", func() {
+		err = errors.New("go-mysql returned an error")
+	})
 	if err != nil {
 		if err != context.Canceled && err != context.DeadlineExceeded {
 			c.Lock()

diff --git a/syncer/syncer.go b/syncer/syncer.go
@@ -1321,6 +1321,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 		}
 
 		if err != nil {
+			// TODO: wrap the error with terror, and attach binlog position
 			tctx.L().Error("fail to fetch binlog", log.ShortError(err))
 
 			if isConnectionRefusedError(err) {
@@ -1346,7 +1347,7 @@ func (s *Syncer) Run(ctx context.Context) (err error) {
 				continue
 			}
 
-			return err
+			return terror.ErrSyncerGetEvent.Generate(err)
 		}
 
 		// time duration for reading an event from relay log or upstream master.

diff --git a/tests/incremental_mode/run.sh b/tests/incremental_mode/run.sh
@@ -152,12 +152,27 @@ function run() {
 	sed -i "s/binlog-pos-placeholder-2/$pos2/g" $WORK_DIR/dm-task.yaml
 	sed -i "s/binlog-gtid-placeholder-2/$gtid2/g" $WORK_DIR/dm-task.yaml
 
+	# test graceful display error
+	export GO_FAILPOINTS="github.com/pingcap/dm/syncer/GetEventError=return()"
+	run_dm_worker $WORK_DIR/worker1 $WORKER1_PORT $cur/conf/dm-worker1.toml
+	check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER1_PORT
+	run_dm_worker $WORK_DIR/worker2 $WORKER2_PORT $cur/conf/dm-worker2.toml
+	check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER2_PORT
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"start-task $WORK_DIR/dm-task.yaml --remove-meta"
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"\"ErrCode\": 36069" 2
+
+	kill_dm_worker
+	check_port_offline $WORKER1_PORT 20
+	check_port_offline $WORKER2_PORT 20
+
 	export GO_FAILPOINTS="github.com/pingcap/dm/syncer/WaitUserCancel=return(8)"
 	run_dm_worker $WORK_DIR/worker1 $WORKER1_PORT $cur/conf/dm-worker1.toml
 	check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER1_PORT
 	run_dm_worker $WORK_DIR/worker2 $WORKER2_PORT $cur/conf/dm-worker2.toml
 	check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER2_PORT
-	dmctl_start_task $WORK_DIR/dm-task.yaml --remove-meta
 
 	# check not specify binlog name could also update active relay log
 	if [ $worker1_run_source_1 -gt 0 ]; then