add a failpoint for test

Signed-off-by: ekexium <[email protected]>

Author: ekexium

txnkv/transaction/2pc.go

@@ -1806,6 +1806,7 @@ func (c *twoPhaseCommitter) execute(ctx context.Context) (err error) {
 			return errors.Errorf("unexpected empty pipelinedStart(%s) or pipelinedEnd(%s)",
 				c.pipelinedCommitInfo.pipelinedStart, c.pipelinedCommitInfo.pipelinedEnd)
 		}
+		util.EvalFailpoint("beforePipelinedCommit")
 		return c.commitFlushedMutations(bo)
 	}
 

txnkv/transaction/pipelined_flush.go

@@ -401,7 +401,7 @@ func (c *twoPhaseCommitter) buildPipelinedResolveHandler(commit bool, resolved *
 	// Use time-based callback control
 	lastCallback := atomic.Value{}
 	lastCallback.Store(time.Now().Add(-minProgressUpdateInterval))
-	
+
 	return func(ctx context.Context, r kv.KeyRange) (rangetask.TaskStat, error) {
 		start := r.StartKey
 		res := rangetask.TaskStat{}
@@ -450,7 +450,7 @@ func (c *twoPhaseCommitter) buildPipelinedResolveHandler(commit bool, resolved *
 			}
 			resolved.Add(1)
 			res.CompletedRegions++
-			
+
 			// Update progress periodically
 			if c.txn.pipelinedProgressCallback != nil {
 				now := time.Now()
@@ -459,8 +459,8 @@ func (c *twoPhaseCommitter) buildPipelinedResolveHandler(commit bool, resolved *
 				if timeSinceLastCallback < 0 {
 					timeSinceLastCallback = minProgressUpdateInterval
 				}
-				if timeSinceLastCallback >= minProgressUpdateInterval && 
-				   lastCallback.CompareAndSwap(last, now) {
+				if timeSinceLastCallback >= minProgressUpdateInterval &&
+					lastCallback.CompareAndSwap(last, now) {
 					resolvedCount := int64(resolved.Load())
 					c.txn.pipelinedProgressCallback(
 						c.startTS,
@@ -470,7 +470,7 @@ func (c *twoPhaseCommitter) buildPipelinedResolveHandler(commit bool, resolved *
 					)
 				}
 			}
-			
+
 			done := loc.EndKey == nil || bytes.Compare(loc.EndKey, r.EndKey) >= 0
 			if done {
 				return res, nil
@@ -544,7 +544,7 @@ func (c *twoPhaseCommitter) resolveFlushedLocks(bo *retry.Backoffer, start, end
 					true,
 				)
 			}
-			
+
 			// wait a while before notifying txn_status_cache to evict the txn,
 			// which tolerates slow followers and avoids the situation that the
 			// txn is evicted before the follower catches up.

txnkv/transaction/txn.go

@@ -208,7 +208,7 @@ type KVTxn struct {
 	pipelinedResolveLockConcurrency int
 	writeThrottleRatio              float64
 	// flushBatchDurationEWMA is read before each flush, and written after each flush => no race
-	flushBatchDurationEWMA ewma.MovingAverage
+	flushBatchDurationEWMA    ewma.MovingAverage
 	pipelinedProgressCallback PipelinedProgressCallbackType
 	// progressRecordInitiated is only updated in the flush process, which is single-threaded. No need to use atomic.
 	progressRecordInitiated bool
@@ -1946,7 +1946,7 @@ type PipelinedDMLStatus int
 
 const (
 	// PipelinedDMLExecuting means the transaction is executing
-	PipelinedDMLExecuting PipelinedDMLStatus = iota	
+	PipelinedDMLExecuting PipelinedDMLStatus = iota
 	// PipelinedDMLRollingBack means the transaction is rolling back
 	PipelinedDMLRollingBack
 	// PipelinedDMLResolvingLocks means the transaction is resolving locks
@@ -1962,5 +1962,5 @@ type PipelinedProgressCallbackType func(startTS uint64, status PipelinedDMLStatu
 
 // SetPipelinedProgressCallback sets the callback function for pipelined DML resolve lock progress
 func (txn *KVTxn) SetPipelinedProgressCallback(callback PipelinedProgressCallbackType) {
-    txn.pipelinedProgressCallback = callback
-}
+	txn.pipelinedProgressCallback = callback
+}