Merge branch 'master' into 3.21.01.0

Author: ffffwh

drivers/mysql/mysql/applier_gtid_executed.go

@@ -252,13 +252,13 @@ func (a *ApplierIncr) cleanGtidExecuted(sid uuid.UUID, txSid string) error {
 	}
 	a.logger.Debug("incr. cleanup after WaitForExecution")
 
-	var intervalStr string
-	{
+	intervalStr := func() string {
 		a.gtidSetLock.RLock()
+		defer a.gtidSetLock.RUnlock()
 		intervals := base.GetIntervals(a.gtidSet, txSid)
-		intervalStr = base.StringInterval(intervals)
-		a.gtidSetLock.RUnlock()
-	}
+		return base.StringInterval(intervals)
+	}()
+
 	// The TX is unnecessary if we first insert and then delete.
 	// However, consider `binlog_group_commit_sync_delay > 0`,
 	// `begin; delete; insert; commit;` (1 TX) is faster than `insert; delete;` (2 TX)

drivers/mysql/mysql/applier_incr.go

@@ -210,15 +210,15 @@ func (a *ApplierIncr) heterogeneousReplay() {
 			txSid := binlogEntry.Coordinates.GetSid()
 
 			gtidSetItem := a.gtidItemMap.GetItem(binlogEntry.Coordinates.SID)
-			{
+			txExecuted := func() bool {
 				a.gtidSetLock.RLock()
+				defer a.gtidSetLock.RUnlock()
 				intervals := base.GetIntervals(a.gtidSet, txSid)
-				if base.IntervalSlicesContainOne(intervals, binlogEntry.Coordinates.GNO) {
-					// entry executed
-					a.logger.Info("skip an executed tx", "sid", txSid, "gno", binlogEntry.Coordinates.GNO)
-					continue
-				}
-				a.gtidSetLock.RUnlock()
+				return base.IntervalSlicesContainOne(intervals, binlogEntry.Coordinates.GNO)
+			}()
+			if txExecuted {
+				a.logger.Info("skip an executed tx", "sid", txSid, "gno", binlogEntry.Coordinates.GNO)
+				continue
 			}
 			// endregion
 			// this must be after duplication check