store/tikv: resolve all locks no matter they're expired or not in BatchResolveLocks

store/tikv/lock_resolver.go

@@ -184,23 +184,11 @@ func (lr *LockResolver) BatchResolveLocks(bo *Backoffer, locks []*Lock, loc Regi
 
 	tikvLockResolverCountWithBatchResolve.Inc()
 
-	expiredLocks := make([]*Lock, 0, len(locks))
-	for _, l := range locks {
-		if lr.store.GetOracle().IsExpired(l.TxnID, l.TTL) {
-			tikvLockResolverCountWithExpired.Inc()
-			expiredLocks = append(expiredLocks, l)
-		} else {
-			tikvLockResolverCountWithNotExpired.Inc()
-		}
-	}
-	if len(expiredLocks) != len(locks) {
-		logutil.BgLogger().Error("BatchResolveLocks: maybe safe point is wrong!",
-			zap.Int("get locks", len(locks)),
-			zap.Int("expired locks", len(expiredLocks)))
-		return false, nil
-	}
+	// The GCWorker kill all ongoing transactions, because it must make sure all
+	// locks have been cleaned before GC.
+	expiredLocks := locks
 
-	startTS, err := lr.store.GetOracle().GetTimestamp(bo.ctx)
+	callerStartTS, err := lr.store.GetOracle().GetTimestamp(bo.ctx)
 	if err != nil {
 		return false, errors.Trace(err)
 	}
@@ -211,19 +199,17 @@ func (lr *LockResolver) BatchResolveLocks(bo *Backoffer, locks []*Lock, loc Regi
 		if _, ok := txnInfos[l.TxnID]; ok {
 			continue
 		}
+		tikvLockResolverCountWithExpired.Inc()
 
-		currentTS, err := lr.store.GetOracle().GetLowResolutionTimestamp(bo.ctx)
-		if err != nil {
-			return false, err
-		}
-		status, err := lr.getTxnStatus(bo, l.TxnID, l.Primary, startTS, currentTS)
+		// Use currentTS = math.MaxUint64 means rollback the txn, no matter the lock is expired or not!
+		status, err := lr.getTxnStatus(bo, l.TxnID, l.Primary, callerStartTS, math.MaxUint64)
 		if err != nil {
 			return false, err
 		}
 
 		if status.ttl > 0 {
-			// Do not clean lock that is not expired.
-			continue
+			logutil.BgLogger().Error("BatchResolveLocks fail to clean locks, this result is not expected!")
+			return false, errors.New("TiDB ask TiKV to rollback locks but it doesn't, the protocol maybe wrong")
 		}
 
 		txnInfos[l.TxnID] = uint64(status.commitTS)

store/tikv/lock_test.go

@@ -386,6 +386,28 @@ func (s *testLockSuite) TestLockTTL(c *C) {
 	s.ttlEquals(c, l.TTL, defaultLockTTL+uint64(time.Since(start)/time.Millisecond))
 }
 
+func (s *testLockSuite) TestBatchResolveLocks(c *C) {
+	txn, err := s.store.Begin()
+	c.Assert(err, IsNil)
+	txn.Set(kv.Key("key"), []byte("value"))
+	s.prewriteTxn(c, txn.(*tikvTxn))
+	l := s.mustGetLock(c, []byte("key"))
+	msBeforeLockExpired := s.store.GetOracle().UntilExpired(l.TxnID, l.TTL)
+	c.Assert(msBeforeLockExpired, Greater, int64(0))
+
+	lr := newLockResolver(s.store)
+	bo := NewBackoffer(context.Background(), GcResolveLockMaxBackoff)
+	loc, err := lr.store.GetRegionCache().LocateKey(bo, l.Primary)
+	c.Assert(err, IsNil)
+	// Check BatchResolveLocks resolve the lock even the ttl is not expired.
+	succ, err := lr.BatchResolveLocks(bo, []*Lock{l}, loc.Region)
+	c.Assert(succ, IsTrue)
+	c.Assert(err, IsNil)
+
+	err = txn.Commit(context.Background())
+	c.Assert(err, NotNil)
+}
+
 func (s *testLockSuite) TestNewLockZeroTTL(c *C) {
 	l := NewLock(&kvrpcpb.LockInfo{})
 	c.Assert(l.TTL, Equals, uint64(0))