fix(mts): retry NOWAIT lock failures and resolve contextcheck lint errors

Under concurrent MTS workers, MySQL 8.0 may return ER_LOCK_NOWAIT (3572)
when gap locks on the ghost table's secondary indexes conflict between
parallel statements. This was not classified as retryable, causing the
MTS worker to abort and cascade a migration failure.

Changes:
- Add errno 3572 (ER_LOCK_NOWAIT) to isRetryableApplyError alongside
  existing 1205 (lock wait timeout) and 1213 (deadlock)
- Remove context.Background() fallback in ApplyDMLEventQueries; all
  callers already pass a valid context (resolves contextcheck lint)
- Use the passed ctx parameter in applyMTSWorkerJob defer instead of
  mgtr.migrationContext.GetContext() (resolves contextcheck lint)

Author: jackiesre721

go/logic/applier.go

@@ -481,17 +481,22 @@ func isDeadlockError(err error) bool {
 }
 
 // isRetryableApplyError reports whether a failed DML apply is a transient
-// concurrency error that should be retried. Under concurrent MTS workers both
-// InnoDB deadlocks (1213) and lock-wait timeouts (1205) are expected: gap locks
-// on the ghost table's secondary indexes cause contention between parallel
-// REPLACE INTO / DELETE FROM statements. Both are resolved by retrying the whole
-// transaction, mirroring MySQL's slave_transaction_retries behaviour.
+// concurrency error that should be retried. Under concurrent MTS workers the
+// following errors are expected: InnoDB deadlocks (1213), lock-wait timeouts
+// (1205), and NOWAIT lock failures (3572). Gap locks on the ghost table's
+// secondary indexes cause contention between parallel statements. All are
+// resolved by retrying the whole transaction, mirroring MySQL's
+// slave_transaction_retries behaviour.
 func isRetryableApplyError(err error) bool {
 	var mysqlErr *drivermysql.MySQLError
 	if !errors.As(err, &mysqlErr) {
 		return false
 	}
-	return mysqlErr.Number == 1213 || mysqlErr.Number == 1205
+	switch mysqlErr.Number {
+	case 1205, 1213, 3572:
+		return true
+	}
+	return false
 }
 
 // retryOnLockWaitTimeout retries the given operation on MySQL lock wait timeout
@@ -1816,9 +1821,6 @@ func (apl *Applier) executeBatchWithWarningChecking(ctx context.Context, tx *gos
 // ApplyDMLEventQueries applies multiple DML queries onto the _ghost_ table
 func (apl *Applier) ApplyDMLEventQueries(ctx context.Context, dmlEvents [](*binlog.BinlogDMLEvent)) error {
 	var totalDelta int64
-	if ctx == nil {
-		ctx = context.Background()
-	}
 
 	err := func() error {
 		conn, err := apl.db.Conn(ctx)

go/logic/migrator.go

@@ -2130,7 +2130,7 @@ func (mgtr *Migrator) applyMTSWorkerJob(ctx context.Context, workerID int, appli
 		if applyErr != nil {
 			mgtr.commitBarrier.commit(job.sequenceNum)
 			mgtr.commitBarrier.completeDelegatedJob()
-			_ = base.SendWithContext(mgtr.migrationContext.GetContext(), mgtr.migrationContext.PanicAbort, applyErr)
+			_ = base.SendWithContext(ctx, mgtr.migrationContext.PanicAbort, applyErr)
 		}
 	}()