func (s *testAsyncCommitSuite) TestAsyncCommitLinearizability() {
	t1 := s.beginAsyncCommitWithLinearizability()
	t2 := s.beginAsyncCommitWithLinearizability()
	err := t1.Set([]byte("a"), []byte("a1"))
	s.Nil(err)
	err = t2.Set([]byte("b"), []byte("b1"))
	s.Nil(err)
	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
	err = t2.Commit(ctx)
	s.Nil(err)
	err = t1.Commit(ctx)
	s.Nil(err)
	commitTS1 := t1.GetCommitTS()
	commitTS2 := t2.GetCommitTS()
	s.Less(commitTS2, commitTS1)
}
It's just a simple validation of linearizability.
Extra tests are needed to test this feature with the control of the TiKV cluster.
package tikv_test
import (
	"bytes"
	"context"
	"fmt"
	"math"
	"sync/atomic"
	"testing"
	"time"
	"github.com/pingcap/errors"
	"github.com/pingcap/kvproto/pkg/kvrpcpb"
	"github.com/pingcap/tidb/store/mockstore/unistore"
	"github.com/stretchr/testify/suite"
	tikverr "github.com/tikv/client-go/v2/error"
	"github.com/tikv/client-go/v2/kv"
	"github.com/tikv/client-go/v2/oracle"
	"github.com/tikv/client-go/v2/testutils"
	"github.com/tikv/client-go/v2/tikv"
	"github.com/tikv/client-go/v2/tikvrpc"
	"github.com/tikv/client-go/v2/txnkv"
	"github.com/tikv/client-go/v2/txnkv/transaction"
	"github.com/tikv/client-go/v2/txnkv/txnlock"
	"github.com/tikv/client-go/v2/util"
)
func TestAsyncCommit(t *testing.T) {
	suite.Run(t, new(testAsyncCommitSuite))
}
type testAsyncCommitCommon struct {
	suite.Suite
	cluster testutils.Cluster
	store   *tikv.KVStore
}
func (s *testAsyncCommitCommon) setUpTest() {
	if *withTiKV {
		s.store = NewTestStore(s.T())
		return
	}
	client, pdClient, cluster, err := unistore.New("")
	s.Require().Nil(err)
	unistore.BootstrapWithSingleStore(cluster)
	s.cluster = cluster
	store, err := tikv.NewTestTiKVStore(fpClient{Client: client}, pdClient, nil, nil, 0)
	s.Require().Nil(err)
	s.store = store
}
func (s *testAsyncCommitCommon) tearDownTest() {
	s.store.Close()
}
func (s *testAsyncCommitCommon) putAlphabets(enableAsyncCommit bool) {
	for ch := byte('a'); ch <= byte('z'); ch++ {
		s.putKV([]byte{ch}, []byte{ch}, enableAsyncCommit)
	}
}
func (s *testAsyncCommitCommon) putKV(key, value []byte, enableAsyncCommit bool) (uint64, uint64) {
	txn := s.beginAsyncCommit()
	err := txn.Set(key, value)
	s.Nil(err)
	err = txn.Commit(context.Background())
	s.Nil(err)
	return txn.StartTS(), txn.GetCommitTS()
}
func (s *testAsyncCommitCommon) mustGetFromTxn(txn transaction.TxnProbe, key, expectedValue []byte) {
	v, err := txn.Get(context.Background(), key)
	s.Nil(err)
	s.Equal(v, expectedValue)
}
func (s *testAsyncCommitCommon) mustGetLock(key []byte) *txnkv.Lock {
	ver, err := s.store.CurrentTimestamp(oracle.GlobalTxnScope)
	s.Nil(err)
	req := tikvrpc.NewRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{
		Key:     key,
		Version: ver,
	})
	bo := tikv.NewBackofferWithVars(context.Background(), 5000, nil)
	loc, err := s.store.GetRegionCache().LocateKey(bo, key)
	s.Nil(err)
	resp, err := s.store.SendReq(bo, req, loc.Region, time.Second*10)
	s.Nil(err)
	s.NotNil(resp.Resp)
	keyErr := resp.Resp.(*kvrpcpb.GetResponse).GetError()
	s.NotNil(keyErr)
	lock, err := txnlock.ExtractLockFromKeyErr(keyErr)
	s.Nil(err)
	return lock
}
func (s *testAsyncCommitCommon) mustPointGet(key, expectedValue []byte) {
	snap := s.store.GetSnapshot(math.MaxUint64)
	value, err := snap.Get(context.Background(), key)
	s.Nil(err)
	s.Equal(value, expectedValue)
}
func (s *testAsyncCommitCommon) mustGetFromSnapshot(version uint64, key, expectedValue []byte) {
	snap := s.store.GetSnapshot(version)
	value, err := snap.Get(context.Background(), key)
	s.Nil(err)
	s.Equal(value, expectedValue)
}
func (s *testAsyncCommitCommon) mustGetNoneFromSnapshot(version uint64, key []byte) {
	snap := s.store.GetSnapshot(version)
	_, err := snap.Get(context.Background(), key)
	s.Equal(errors.Cause(err), tikverr.ErrNotExist)
}
func (s *testAsyncCommitCommon) beginAsyncCommitWithLinearizability() transaction.TxnProbe {
	txn := s.beginAsyncCommit()
	txn.SetCausalConsistency(false)
	return txn
}
func (s *testAsyncCommitCommon) beginAsyncCommit() transaction.TxnProbe {
	txn, err := s.store.Begin()
	s.Nil(err)
	txn.SetEnableAsyncCommit(true)
	return transaction.TxnProbe{KVTxn: txn}
}
func (s *testAsyncCommitCommon) begin() transaction.TxnProbe {
	txn, err := s.store.Begin()
	s.Nil(err)
	return transaction.TxnProbe{KVTxn: txn}
}
func (s *testAsyncCommitCommon) begin1PC() transaction.TxnProbe {
	txn, err := s.store.Begin()
	s.Nil(err)
	txn.SetEnable1PC(true)
	return transaction.TxnProbe{KVTxn: txn}
}
type testAsyncCommitSuite struct {
	testAsyncCommitCommon
	bo *tikv.Backoffer
}
func (s *testAsyncCommitSuite) SetupTest() {
	s.testAsyncCommitCommon.setUpTest()
	s.bo = tikv.NewBackofferWithVars(context.Background(), 5000, nil)
}
func (s *testAsyncCommitSuite) TearDownTest() {
	s.testAsyncCommitCommon.tearDownTest()
}
func (s *testAsyncCommitSuite) lockKeysWithAsyncCommit(keys, values [][]byte, primaryKey, primaryValue []byte, commitPrimary bool) (uint64, uint64) {
	txn, err := s.store.Begin()
	s.Nil(err)
	txn.SetEnableAsyncCommit(true)
	for i, k := range keys {
		if len(values[i]) > 0 {
			err = txn.Set(k, values[i])
		} else {
			err = txn.Delete(k)
		}
		s.Nil(err)
	}
	if len(primaryValue) > 0 {
		err = txn.Set(primaryKey, primaryValue)
	} else {
		err = txn.Delete(primaryKey)
	}
	s.Nil(err)
	txnProbe := transaction.TxnProbe{KVTxn: txn}
	tpc, err := txnProbe.NewCommitter(0)
	s.Nil(err)
	tpc.SetPrimaryKey(primaryKey)
	tpc.SetUseAsyncCommit()
	ctx := context.Background()
	err = tpc.PrewriteAllMutations(ctx)
	s.Nil(err)
	if commitPrimary {
		commitTS, err := s.store.GetOracle().GetTimestamp(ctx, &oracle.Option{TxnScope: oracle.GlobalTxnScope})
		s.Nil(err)
		tpc.SetCommitTS(commitTS)
		err = tpc.CommitMutations(ctx)
		s.Nil(err)
	}
	return txn.StartTS(), tpc.GetCommitTS()
}
func (s *testAsyncCommitSuite) TestCheckSecondaries() {
	if *withTiKV {
		return
	}
	s.putAlphabets(true)
	loc, err := s.store.GetRegionCache().LocateKey(s.bo, []byte("a"))
	s.Nil(err)
	newRegionID, peerID := s.cluster.AllocID(), s.cluster.AllocID()
	s.cluster.Split(loc.Region.GetID(), newRegionID, []byte("e"), []uint64{peerID}, peerID)
	s.store.GetRegionCache().InvalidateCachedRegion(loc.Region)
	s.lockKeysWithAsyncCommit([][]byte{}, [][]byte{}, []byte("z"), []byte("z"), false)
	lock := s.mustGetLock([]byte("z"))
	lock.UseAsyncCommit = true
	ts, err := s.store.GetOracle().GetTimestamp(context.Background(), &oracle.Option{TxnScope: oracle.GlobalTxnScope})
	s.Nil(err)
	var lockutil txnlock.LockProbe
	status := lockutil.NewLockStatus(nil, true, ts)
	resolver := tikv.NewLockResolverProb(s.store.GetLockResolver())
	err = resolver.ResolveLockAsync(s.bo, lock, status)
	s.Nil(err)
	currentTS, err := s.store.GetOracle().GetTimestamp(context.Background(), &oracle.Option{TxnScope: oracle.GlobalTxnScope})
	s.Nil(err)
	status, err = resolver.GetTxnStatus(s.bo, lock.TxnID, []byte("z"), currentTS, currentTS, true, false, nil)
	s.Nil(err)
	s.True(status.IsCommitted())
	s.Equal(status.CommitTS(), ts)
	ts, err = s.store.GetOracle().GetTimestamp(context.Background(), &oracle.Option{TxnScope: oracle.GlobalTxnScope})
	s.Nil(err)
	commitTs := ts + 10
	gotCheckA := int64(0)
	gotCheckB := int64(0)
	gotResolve := int64(0)
	gotOther := int64(0)
	mock := mockResolveClient{
		inner: s.store.GetTiKVClient(),
		onCheckSecondaries: func(req *kvrpcpb.CheckSecondaryLocksRequest) (*tikvrpc.Response, error) {
			if req.StartVersion != ts {
				return nil, errors.Errorf("Bad start version: %d, expected: %d", req.StartVersion, ts)
			}
			var resp kvrpcpb.CheckSecondaryLocksResponse
			for _, k := range req.Keys {
				if bytes.Equal(k, []byte("a")) {
					atomic.StoreInt64(&gotCheckA, 1)
					resp = kvrpcpb.CheckSecondaryLocksResponse{
						Locks:    []*kvrpcpb.LockInfo{{Key: []byte("a"), PrimaryLock: []byte("z"), LockVersion: ts, UseAsyncCommit: true}},
						CommitTs: commitTs,
					}
				} else if bytes.Equal(k, []byte("i")) {
					atomic.StoreInt64(&gotCheckB, 1)
					resp = kvrpcpb.CheckSecondaryLocksResponse{
						Locks:    []*kvrpcpb.LockInfo{},
						CommitTs: commitTs,
					}
				} else {
					fmt.Printf("Got other key: %s\n", k)
					atomic.StoreInt64(&gotOther, 1)
				}
			}
			return &tikvrpc.Response{Resp: &resp}, nil
		},
		onResolveLock: func(req *kvrpcpb.ResolveLockRequest) (*tikvrpc.Response, error) {
			if req.StartVersion != ts {
				return nil, errors.Errorf("Bad start version: %d, expected: %d", req.StartVersion, ts)
			}
			if req.CommitVersion != commitTs {
				return nil, errors.Errorf("Bad commit version: %d, expected: %d", req.CommitVersion, commitTs)
			}
			for _, k := range req.Keys {
				if bytes.Equal(k, []byte("a")) || bytes.Equal(k, []byte("z")) {
					atomic.StoreInt64(&gotResolve, 1)
				} else {
					atomic.StoreInt64(&gotOther, 1)
				}
			}
			resp := kvrpcpb.ResolveLockResponse{}
			return &tikvrpc.Response{Resp: &resp}, nil
		},
	}
	s.store.SetTiKVClient(&mock)
	status = lockutil.NewLockStatus([][]byte{[]byte("a"), []byte("i")}, true, 0)
	lock = &txnkv.Lock{
		Key:            []byte("a"),
		Primary:        []byte("z"),
		TxnID:          ts,
		LockType:       kvrpcpb.Op_Put,
		UseAsyncCommit: true,
		MinCommitTS:    ts + 5,
	}
	_ = s.beginAsyncCommit()
	err = resolver.ResolveLockAsync(s.bo, lock, status)
	s.Nil(err)
	s.Equal(gotCheckA, int64(1))
	s.Equal(gotCheckB, int64(1))
	s.Equal(gotOther, int64(0))
	s.Equal(gotResolve, int64(1))
	ts, err = s.store.GetOracle().GetTimestamp(context.Background(), &oracle.Option{TxnScope: oracle.GlobalTxnScope})
	s.Nil(err)
	commitTs = ts + 10
	gotCheckA = int64(0)
	gotCheckB = int64(0)
	gotResolve = int64(0)
	gotOther = int64(0)
	mock.onResolveLock = func(req *kvrpcpb.ResolveLockRequest) (*tikvrpc.Response, error) {
		if req.StartVersion != ts {
			return nil, errors.Errorf("Bad start version: %d, expected: %d", req.StartVersion, ts)
		}
		if req.CommitVersion != commitTs {
			return nil, errors.Errorf("Bad commit version: %d, expected: 0", req.CommitVersion)
		}
		for _, k := range req.Keys {
			if bytes.Equal(k, []byte("a")) || bytes.Equal(k, []byte("z")) {
				atomic.StoreInt64(&gotResolve, 1)
			} else {
				atomic.StoreInt64(&gotOther, 1)
			}
		}
		resp := kvrpcpb.ResolveLockResponse{}
		return &tikvrpc.Response{Resp: &resp}, nil
	}
	lock.TxnID = ts
	lock.MinCommitTS = ts + 5
	err = resolver.ResolveLockAsync(s.bo, lock, status)
	s.Nil(err)
	s.Equal(gotCheckA, int64(1))
	s.Equal(gotCheckB, int64(1))
	s.Equal(gotResolve, int64(1))
	s.Equal(gotOther, int64(0))
}
func (s *testAsyncCommitSuite) TestRepeatableRead() {
	var sessionID uint64 = 0
	test := func(isPessimistic bool) {
		s.putKV([]byte("k1"), []byte("v1"), true)
		sessionID++
		ctx := context.WithValue(context.Background(), util.SessionID, sessionID)
		txn1 := s.beginAsyncCommit()
		txn1.SetPessimistic(isPessimistic)
		s.mustGetFromTxn(txn1, []byte("k1"), []byte("v1"))
		txn1.Set([]byte("k1"), []byte("v2"))
		for i := 0; i < 20; i++ {
			_, err := s.store.GetOracle().GetTimestamp(ctx, &oracle.Option{TxnScope: oracle.GlobalTxnScope})
			s.Nil(err)
		}
		txn2 := s.beginAsyncCommit()
		s.mustGetFromTxn(txn2, []byte("k1"), []byte("v1"))
		err := txn1.Commit(ctx)
		s.Nil(err)
		s.True(txn1.IsAsyncCommit())
		s.mustGetFromTxn(txn2, []byte("k1"), []byte("v1"))
		err = txn2.Rollback()
		s.Nil(err)
		txn3 := s.beginAsyncCommit()
		s.mustGetFromTxn(txn3, []byte("k1"), []byte("v2"))
		err = txn3.Rollback()
		s.Nil(err)
	}
	test(false)
	test(true)
}