This is an automated cherry-pick of pingcap#11251

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

cdc/sink/dmlsink/txn/worker.go

@@ -46,6 +46,8 @@ type worker struct {
 	flushInterval            time.Duration
 	hasPending               bool
 	postTxnExecutedCallbacks []func()
+
+	lastSlowConflictDetectLog map[model.TableID]time.Time
 }
 
 func newWorker(ctx context.Context, changefeedID model.ChangeFeedID,
@@ -70,6 +72,8 @@ func newWorker(ctx context.Context, changefeedID model.ChangeFeedID,
 		flushInterval:            backend.MaxFlushInterval(),
 		hasPending:               false,
 		postTxnExecutedCallbacks: make([]func(), 0, 1024),
+
+		lastSlowConflictDetectLog: make(map[model.TableID]time.Time),
 	}
 }
 
@@ -87,6 +91,7 @@ func (w *worker) runLoop(txnCh <-chan causality.TxnWithNotifier[*txnEvent]) erro
 		zap.String("changefeedID", w.changefeed),
 		zap.Int("workerID", w.ID))
 
+<<<<<<< HEAD
 	ticker := time.NewTicker(w.flushInterval)
 	defer ticker.Stop()
 
@@ -95,13 +100,28 @@ func (w *worker) runLoop(txnCh <-chan causality.TxnWithNotifier[*txnEvent]) erro
 	overseerTicker := time.NewTicker(time.Second)
 	defer overseerTicker.Stop()
 	startToWork := time.Now()
+=======
+	cleanSlowLogHistory := time.NewTicker(time.Hour)
+	defer cleanSlowLogHistory.Stop()
+
+	start := time.Now()
+>>>>>>> d662c77362 (cdc: log slow conflict detect every 60s (#11251))
 	for {
 		select {
 		case <-w.ctx.Done():
 			log.Info("Transaction dmlSink worker exits as canceled",
 				zap.String("changefeedID", w.changefeed),
 				zap.Int("workerID", w.ID))
 			return nil
+		case <-cleanSlowLogHistory.C:
+			lastSlowConflictDetectLog := w.lastSlowConflictDetectLog
+			w.lastSlowConflictDetectLog = make(map[model.TableID]time.Time)
+			now := time.Now()
+			for tableID, lastLog := range lastSlowConflictDetectLog {
+				if now.Sub(lastLog) <= time.Minute {
+					w.lastSlowConflictDetectLog[tableID] = lastLog
+				}
+			}
 		case txn := <-txnCh:
 			if txn.TxnEvent != nil {
 				needFlush = w.onEvent(txn.TxnEvent, txn.PostTxnExecuted)
@@ -142,8 +162,24 @@ func (w *worker) onEvent(txn *txnEvent, postTxnExecuted func()) bool {
 		return false
 	}
 
-	w.metricConflictDetectDuration.Observe(txn.conflictResolved.Sub(txn.start).Seconds())
+	conflictDetectTime := txn.conflictResolved.Sub(txn.start).Seconds()
+	w.metricConflictDetectDuration.Observe(conflictDetectTime)
 	w.metricQueueDuration.Observe(time.Since(txn.start).Seconds())
+
+	// Log tables which conflict detect time larger than 1 minute.
+	if conflictDetectTime > float64(60) {
+		now := time.Now()
+		// Log slow conflict detect tables every minute.
+		if lastLog, ok := w.lastSlowConflictDetectLog[txn.Event.PhysicalTableID]; !ok || now.Sub(lastLog) > time.Minute {
+			log.Warn("Transaction dmlSink finds a slow transaction in conflict detector",
+				zap.String("changefeedID", w.changefeed),
+				zap.Int("workerID", w.ID),
+				zap.Int64("TableID", txn.Event.PhysicalTableID),
+				zap.Float64("seconds", conflictDetectTime))
+			w.lastSlowConflictDetectLog[txn.Event.PhysicalTableID] = now
+		}
+	}
+
 	w.metricTxnWorkerHandledRows.Add(float64(len(txn.Event.Rows)))
 	w.postTxnExecutedCallbacks = append(w.postTxnExecutedCallbacks, postTxnExecuted)
 	return w.backend.OnTxnEvent(txn.TxnCallbackableEvent)