dependency: upgrade TiDB dependency (pingcap#4310)

close pingcap#4298

dm/dm/portal/api_test.go

@@ -22,19 +22,16 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"sort"
-	"strconv"
-	"strings"
 	"testing"
 
 	sqlmock "github.com/DATA-DOG/go-sqlmock"
-	"github.com/go-sql-driver/mysql"
 	. "github.com/pingcap/check"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb-tools/pkg/filter"
 	router "github.com/pingcap/tidb-tools/pkg/table-router"
-	"github.com/pingcap/tidb/br/pkg/mock"
 
 	"github.com/pingcap/tiflow/dm/dm/config"
+	"github.com/pingcap/tiflow/dm/pkg/conn"
 )
 
 var _ = Suite(&testPortalSuite{})
@@ -46,7 +43,7 @@ type testPortalSuite struct {
 
 	taskConfig *DMTaskConfig
 
-	mockCluster     *mock.Cluster
+	mockCluster     *conn.Cluster
 	mockClusterPort int
 }
 
@@ -73,14 +70,11 @@ func (t *testPortalSuite) SetUpSuite(c *C) {
 	}
 
 	t.initTaskCfg()
-	cluster, err := mock.NewCluster()
+	cluster, err := conn.NewCluster()
 	c.Assert(err, IsNil)
 	t.mockCluster = cluster
 	c.Assert(t.mockCluster.Start(), IsNil)
-	config, err := mysql.ParseDSN(cluster.DSN)
-	c.Assert(err, IsNil)
-	t.mockClusterPort, err = strconv.Atoi(strings.Split(config.Addr, ":")[1])
-	c.Assert(err, IsNil)
+	t.mockClusterPort = cluster.Port
 }
 
 func (t *testPortalSuite) TearDownSuite(c *C) {

dm/pkg/conn/mockdb.go

@@ -15,9 +15,21 @@ package conn
 
 import (
 	"database/sql"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"time"
 
 	"github.com/DATA-DOG/go-sqlmock"
 	check "github.com/pingcap/check"
+	tidbConfig "github.com/pingcap/tidb/config"
+	"github.com/pingcap/tidb/domain"
+	"github.com/pingcap/tidb/kv"
+	"github.com/pingcap/tidb/server"
+	"github.com/pingcap/tidb/session"
+	"github.com/pingcap/tidb/store/mockstore"
+	"github.com/tikv/client-go/v2/testutils"
 
 	"github.com/pingcap/tiflow/dm/dm/config"
 )
@@ -61,3 +73,106 @@ func InitVersionDB(c *check.C) sqlmock.Sqlmock {
 	}
 	return mock
 }
+
+// TODO: export Config in https://github.com/pingcap/tidb/blob/a8fa29b56d633b1ec843e21cb89131dd4fd601db/br/pkg/mock/mock_cluster.go#L35
+// Cluster is mock tidb cluster.
+type Cluster struct {
+	*server.Server
+	testutils.Cluster
+	kv.Storage
+	*server.TiDBDriver
+	*domain.Domain
+	Port int
+}
+
+// NewCluster create a new mock cluster.
+func NewCluster() (*Cluster, error) {
+	cluster := &Cluster{}
+
+	storage, err := mockstore.NewMockStore(
+		mockstore.WithClusterInspector(func(c testutils.Cluster) {
+			mockstore.BootstrapWithSingleStore(c)
+			cluster.Cluster = c
+		}),
+	)
+	if err != nil {
+		return nil, err
+	}
+	cluster.Storage = storage
+
+	session.SetSchemaLease(0)
+	session.DisableStats4Test()
+	dom, err := session.BootstrapSession(storage)
+	if err != nil {
+		return nil, err
+	}
+	cluster.Domain = dom
+
+	return cluster, nil
+}
+
+// Start runs a mock cluster.
+func (mock *Cluster) Start() error {
+	// choose a random available port
+	l1, _ := net.Listen("tcp", "127.0.0.1:")
+	statusPort := l1.Addr().(*net.TCPAddr).Port
+
+	// choose a random available port
+	l2, _ := net.Listen("tcp", "127.0.0.1:")
+	addrPort := l2.Addr().(*net.TCPAddr).Port
+
+	mock.TiDBDriver = server.NewTiDBDriver(mock.Storage)
+	cfg := tidbConfig.NewConfig()
+	cfg.Port = uint(addrPort)
+	cfg.Store = "tikv"
+	cfg.Status.StatusPort = uint(statusPort)
+	cfg.Status.ReportStatus = true
+	cfg.Socket = fmt.Sprintf("/tmp/tidb-mock-%d.sock", time.Now().UnixNano())
+
+	// close port for next listen in NewServer
+	l1.Close()
+	l2.Close()
+	svr, err := server.NewServer(cfg, mock.TiDBDriver)
+	if err != nil {
+		return err
+	}
+	mock.Server = svr
+	go func() {
+		if err1 := svr.Run(); err1 != nil {
+			panic(err1)
+		}
+	}()
+	waitUntilServerOnline(cfg.Status.StatusPort)
+	mock.Port = addrPort
+	return nil
+}
+
+// Stop stops a mock cluster.
+func (mock *Cluster) Stop() {
+	if mock.Domain != nil {
+		mock.Domain.Close()
+	}
+	if mock.Storage != nil {
+		_ = mock.Storage.Close()
+	}
+	if mock.Server != nil {
+		mock.Server.Close()
+	}
+}
+
+func waitUntilServerOnline(statusPort uint) {
+	// connect http status
+	statusURL := fmt.Sprintf("http://127.0.0.1:%d/status", statusPort)
+	for retry := 0; retry < 100; retry++ {
+		// nolint:gosec,noctx
+		// #nosec G107
+		resp, err := http.Get(statusURL)
+		if err == nil {
+			// Ignore errors.
+			_, _ = io.ReadAll(resp.Body)
+			_ = resp.Body.Close()
+			break
+		}
+		time.Sleep(time.Millisecond * 10)
+	}
+}

dm/pkg/schema/tracker.go

@@ -367,7 +367,7 @@ func (tr *Tracker) CreateTableIfNotExists(table *filter.Table, ti *model.TableIn
 	tableName := model.NewCIStr(table.Name)
 	ti = cloneTableInfo(ti)
 	ti.Name = tableName
-	return tr.dom.DDL().CreateTableWithInfo(tr.se, schemaName, ti, ddl.OnExistIgnore, false)
+	return tr.dom.DDL().CreateTableWithInfo(tr.se, schemaName, ti, ddl.OnExistIgnore)
 }
 
 // GetSystemVar gets a variable from schema tracker.

dm/syncer/ddl_test.go

@@ -17,20 +17,18 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strconv"
 	"strings"
 
 	"github.com/DATA-DOG/go-sqlmock"
-	"github.com/go-sql-driver/mysql"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/tidb-tools/pkg/filter"
 	router "github.com/pingcap/tidb-tools/pkg/table-router"
-	"github.com/pingcap/tidb/br/pkg/mock"
 	"github.com/pingcap/tidb/parser"
 	"github.com/pingcap/tidb/parser/ast"
 	"go.uber.org/zap"
 
 	"github.com/pingcap/tiflow/dm/dm/config"
+	"github.com/pingcap/tiflow/dm/pkg/conn"
 	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	parserpkg "github.com/pingcap/tiflow/dm/pkg/parser"
@@ -446,13 +444,10 @@ func (s *testDDLSuite) TestResolveOnlineDDL(c *C) {
 	p := parser.New()
 
 	ec := &eventContext{tctx: tctx}
-	cluster, err := mock.NewCluster()
+	cluster, err := conn.NewCluster()
 	c.Assert(err, IsNil)
 	c.Assert(cluster.Start(), IsNil)
-	mysqlConfig, err := mysql.ParseDSN(cluster.DSN)
-	c.Assert(err, IsNil)
-	mockClusterPort, err := strconv.Atoi(strings.Split(mysqlConfig.Addr, ":")[1])
-	c.Assert(err, IsNil)
+	mockClusterPort := cluster.Port
 	dbCfg := config.GetDBConfigForTest()
 	dbCfg.Port = mockClusterPort
 	dbCfg.Password = ""
@@ -529,12 +524,10 @@ func (s *testDDLSuite) TestMistakeOnlineDDLRegex(c *C) {
 	p := parser.New()
 
 	ec := eventContext{tctx: tctx}
-	cluster, err := mock.NewCluster()
+	cluster, err := conn.NewCluster()
 	c.Assert(err, IsNil)
 	c.Assert(cluster.Start(), IsNil)
-	mysqlConfig, err := mysql.ParseDSN(cluster.DSN)
-	c.Assert(err, IsNil)
-	mockClusterPort, err := strconv.Atoi(strings.Split(mysqlConfig.Addr, ":")[1])
+	mockClusterPort := cluster.Port
 	c.Assert(err, IsNil)
 	dbCfg := config.GetDBConfigForTest()
 	dbCfg.Port = mockClusterPort

go.mod

@@ -52,13 +52,13 @@ require (
 	github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
 	github.com/philhofer/fwd v1.0.0 // indirect
 	github.com/pingcap/check v0.0.0-20200212061837-5e12011dc712
-	github.com/pingcap/errors v0.11.5-0.20211009033009-93128226aaa3
+	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c
 	github.com/pingcap/failpoint v0.0.0-20210316064728-7acb0f0a3dfd
-	github.com/pingcap/kvproto v0.0.0-20211202065422-a412f7a319c3
+	github.com/pingcap/kvproto v0.0.0-20211207042851-78a55fb8e69c
 	github.com/pingcap/log v0.0.0-20211207084639-71a2e5860834
-	github.com/pingcap/tidb v1.1.0-beta.0.20211229105350-1e7f0dcc63b9
+	github.com/pingcap/tidb v1.1.0-beta.0.20220112050342-d088e3d6fc6c
 	github.com/pingcap/tidb-tools v5.2.3-0.20211105044302-2dabb6641a6e+incompatible
-	github.com/pingcap/tidb/parser v0.0.0-20211229105350-1e7f0dcc63b9
+	github.com/pingcap/tidb/parser v0.0.0-20220112091742-82a75542e83f
 	github.com/prometheus/client_golang v1.7.1
 	github.com/prometheus/client_model v0.2.0
 	github.com/r3labs/diff v1.1.0
@@ -74,7 +74,7 @@ require (
 	github.com/syndtr/goleveldb v1.0.1-0.20210305035536-64b5b1c73954
 	github.com/tidwall/gjson v1.9.1
 	github.com/tidwall/sjson v1.2.2
-	github.com/tikv/client-go/v2 v2.0.0-rc.0.20211223062159-300275dee63e
+	github.com/tikv/client-go/v2 v2.0.0-rc.0.20211229051614-62d6b4a2e8f7
 	github.com/tikv/pd v1.1.0-beta.0.20211118054146-02848d2660ee
 	github.com/tinylib/msgp v1.1.0
 	github.com/uber-go/atomic v1.4.0