sync unit: disable retry for binlog syncer (#635)

* disable retry for binlog syncer

* update reConnect to retryCount

Co-authored-by: Chunzhu Li <lichunzhu@stu.xjtu.edu.cn>

feat: update

pkg/binlog/common/replication.go

@@ -20,18 +20,18 @@ import (
 )
 
 var (
-	// max reconnection times for binlog syncer in go-mysql
-	maxBinlogSyncerReconnect = 60
+	// MaxBinlogSyncerReconnect is the max reconnection times for binlog syncer in go-mysql
+	MaxBinlogSyncerReconnect = 60
 	// SlaveReadTimeout is slave read binlog data timeout, ref: https://dev.mysql.com/doc/refman/8.0/en/replication-options-slave.html#sysvar_slave_net_timeout
 	SlaveReadTimeout      = 1 * time.Minute
 	masterHeartbeatPeriod = 30 * time.Second // master server send heartbeat period: ref: `MASTER_HEARTBEAT_PERIOD` in https://dev.mysql.com/doc/refman/8.0/en/change-master-to.html
 )
 
 // SetDefaultReplicationCfg sets some default value for BinlogSyncerConfig
-func SetDefaultReplicationCfg(cfg *replication.BinlogSyncerConfig) {
+func SetDefaultReplicationCfg(cfg *replication.BinlogSyncerConfig, retryCount int) {
 	cfg.UseDecimal = true // must set true. ref: https://github.com/pingcap/tidb-enterprise-tools/pull/272
 	cfg.VerifyChecksum = true
-	cfg.MaxReconnectAttempts = maxBinlogSyncerReconnect
+	cfg.MaxReconnectAttempts = retryCount
 	cfg.ReadTimeout = SlaveReadTimeout
 	cfg.HeartbeatPeriod = masterHeartbeatPeriod
 }

pkg/utils/encrypt.go

@@ -46,14 +46,9 @@ func Decrypt(ciphertextB64 string) (string, error) {
 
 // DecryptOrPlaintext tries to decrypt base64 encoded ciphertext to plaintext or return plaintext
 func DecryptOrPlaintext(ciphertextB64 string) (string, error) {
-	ciphertext, err := base64.StdEncoding.DecodeString(ciphertextB64)
+	plaintext, err := Decrypt(ciphertextB64)
 	if err != nil {
 		return ciphertextB64, nil
 	}
-
-	plaintext, err := encrypt.Decrypt(ciphertext)
-	if err != nil {
-		return "", terror.Annotatef(err, "can not decrypt password %s", ciphertextB64)
-	}
-	return string(plaintext), nil
+	return plaintext, nil
 }

relay/relay.go

@@ -125,7 +125,7 @@ func NewRealRelay(cfg *Config) Process {
 		Password: cfg.From.Password,
 		Charset:  cfg.Charset,
 	}
-	common.SetDefaultReplicationCfg(&syncerCfg)
+	common.SetDefaultReplicationCfg(&syncerCfg, common.MaxBinlogSyncerReconnect)
 
 	if !cfg.EnableGTID {
 		// for rawMode(true), we only parse FormatDescriptionEvent and RotateEvent
@@ -803,7 +803,7 @@ func (r *Relay) Reload(newCfg *Config) error {
 		Password: newCfg.From.Password,
 		Charset:  newCfg.Charset,
 	}
-	common.SetDefaultReplicationCfg(&syncerCfg)
+	common.SetDefaultReplicationCfg(&syncerCfg, common.MaxBinlogSyncerReconnect)
 
 	if !newCfg.EnableGTID {
 		// for rawMode(true), we only parse FormatDescriptionEvent and RotateEvent

syncer/syncer.go

@@ -2583,7 +2583,10 @@ func (s *Syncer) setSyncCfg() {
 		Password:                s.cfg.From.Password,
 		TimestampStringLocation: s.timezone,
 	}
-	common.SetDefaultReplicationCfg(&syncCfg)
+	// when retry count > 1, go-mysql will retry sync from the previous GTID set in GTID mode,
+	// which may get duplicate binlog event after retry success. so just set retry count = 1, and task
+	// will exit when meet error, and then auto resume by DM itself.
+	common.SetDefaultReplicationCfg(&syncCfg, 1)
 	s.syncCfg = syncCfg
 }