*(both): update dep of etcd, go-mysql and others

cdc/api/owner.go

@@ -24,13 +24,14 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"github.com/tikv/client-go/v2/oracle"
+	"go.etcd.io/etcd/client/v3/concurrency"
+	"go.uber.org/zap"
+
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/cdc/model"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/logutil"
-	"github.com/tikv/client-go/v2/oracle"
-	"go.etcd.io/etcd/clientv3/concurrency"
-	"go.uber.org/zap"
 )
 
 const (

cdc/api/owner_test.go

@@ -23,7 +23,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/stretchr/testify/require"
-	"go.etcd.io/etcd/clientv3/concurrency"
+	"go.etcd.io/etcd/client/v3/concurrency"
 )
 
 func TestHTTPStatus(t *testing.T) {

cdc/api/status.go

@@ -24,7 +24,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/pkg/etcd"
 	"github.com/pingcap/tiflow/pkg/version"
-	"go.etcd.io/etcd/clientv3"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 // status of cdc server

cdc/capture/capture.go

@@ -25,6 +25,13 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	tidbkv "github.com/pingcap/tidb/kv"
+	"github.com/tikv/client-go/v2/tikv"
+	pd "github.com/tikv/pd/client"
+	"go.etcd.io/etcd/client/v3/concurrency"
+	"go.etcd.io/etcd/server/v3/mvcc"
+	"go.uber.org/zap"
+	"golang.org/x/time/rate"
+
 	"github.com/pingcap/tiflow/cdc/kv"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/owner"
@@ -39,12 +46,6 @@ import (
 	"github.com/pingcap/tiflow/pkg/p2p"
 	"github.com/pingcap/tiflow/pkg/pdtime"
 	"github.com/pingcap/tiflow/pkg/version"
-	"github.com/tikv/client-go/v2/tikv"
-	pd "github.com/tikv/pd/client"
-	"go.etcd.io/etcd/clientv3/concurrency"
-	"go.etcd.io/etcd/mvcc"
-	"go.uber.org/zap"
-	"golang.org/x/time/rate"
 )
 
 // Capture represents a Capture server, it monitors the changefeed information in etcd and schedules Task on it.

cdc/processor/agent.go

@@ -22,6 +22,10 @@ import (
 	"github.com/benbjohnson/clock"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"go.etcd.io/etcd/client/v3/concurrency"
+	"go.uber.org/zap"
+	"golang.org/x/time/rate"
+
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/scheduler"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -30,9 +34,6 @@ import (
 	"github.com/pingcap/tiflow/pkg/etcd"
 	"github.com/pingcap/tiflow/pkg/p2p"
 	"github.com/pingcap/tiflow/pkg/version"
-	"go.etcd.io/etcd/clientv3/concurrency"
-	"go.uber.org/zap"
-	"golang.org/x/time/rate"
 )
 
 const (

cdc/processor/agent_test.go

@@ -19,16 +19,17 @@ import (
 	"time"
 
 	"github.com/pingcap/failpoint"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
+
 	"github.com/pingcap/tiflow/cdc/model"
 	pscheduler "github.com/pingcap/tiflow/cdc/scheduler"
 	cdcContext "github.com/pingcap/tiflow/pkg/context"
 	"github.com/pingcap/tiflow/pkg/etcd"
 	"github.com/pingcap/tiflow/pkg/p2p"
 	"github.com/pingcap/tiflow/pkg/version"
-	"github.com/stretchr/testify/mock"
-	"github.com/stretchr/testify/require"
-	"go.etcd.io/etcd/clientv3"
-	"go.etcd.io/etcd/mvcc/mvccpb"
 )
 
 const (

cdc/redo/writer/file.go

@@ -29,13 +29,14 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/br/pkg/storage"
-	"github.com/pingcap/tiflow/cdc/redo/common"
-	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/uber-go/atomic"
-	pioutil "go.etcd.io/etcd/pkg/ioutil"
+	pioutil "go.etcd.io/etcd/pkg/v3/ioutil"
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
+
+	"github.com/pingcap/tiflow/cdc/redo/common"
+	cerror "github.com/pingcap/tiflow/pkg/errors"
 )
 
 const (

cdc/server.go

@@ -28,6 +28,16 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	tidbkv "github.com/pingcap/tidb/kv"
+	"github.com/prometheus/client_golang/prometheus"
+	pd "github.com/tikv/pd/client"
+	"go.etcd.io/etcd/client/pkg/v3/logutil"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/backoff"
+
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/cdc/kv"
 	"github.com/pingcap/tiflow/cdc/sorter/unified"
@@ -41,15 +51,6 @@ import (
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/pingcap/tiflow/pkg/version"
 	p2pProto "github.com/pingcap/tiflow/proto/p2p"
-	"github.com/prometheus/client_golang/prometheus"
-	pd "github.com/tikv/pd/client"
-	"go.etcd.io/etcd/clientv3"
-	"go.etcd.io/etcd/pkg/logutil"
-	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
-	"golang.org/x/sync/errgroup"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/backoff"
 )
 
 const (

cdc/server_test.go

@@ -29,6 +29,12 @@ import (
 	"time"
 
 	"github.com/pingcap/tidb/br/pkg/httputil"
+	"github.com/stretchr/testify/require"
+	"github.com/tikv/pd/pkg/tempurl"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/server/v3/embed"
+	"golang.org/x/sync/errgroup"
+
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -37,11 +43,6 @@ import (
 	"github.com/pingcap/tiflow/pkg/retry"
 	security2 "github.com/pingcap/tiflow/pkg/security"
 	"github.com/pingcap/tiflow/pkg/util"
-	"github.com/stretchr/testify/require"
-	"github.com/tikv/pd/pkg/tempurl"
-	"go.etcd.io/etcd/clientv3"
-	"go.etcd.io/etcd/embed"
-	"golang.org/x/sync/errgroup"
 )
 
 type testServer struct {

dm/dm/config/source_config.go

@@ -47,11 +47,11 @@ const (
 
 var getAllServerIDFunc = utils.GetAllServerID
 
-// SampleConfigFile is sample config file of source.
+// SampleSourceConfig is sample config file of source.
 // The embed source.yaml is a copy of dm/master/source.yaml, because embed
 // can only match regular files in the current directory and subdirectories.
 //go:embed source.yaml
-var SampleConfigFile string
+var SampleSourceConfig string
 
 // PurgeConfig is the configuration for Purger.
 type PurgeConfig struct {
@@ -170,6 +170,18 @@ func ParseYaml(content string) (*SourceConfig, error) {
 	return c, nil
 }
 
+// ParseYamlAndVerify does ParseYaml and Verify.
+func ParseYamlAndVerify(content string) (*SourceConfig, error) {
+	c, err := ParseYaml(content)
+	if err != nil {
+		return nil, err
+	}
+	if err = c.Verify(); err != nil {
+		return nil, err
+	}
+	return c, nil
+}
+
 // EncodeToml encodes config.
 func (c *SourceConfig) EncodeToml() (string, error) {
 	buf := new(bytes.Buffer)
@@ -348,19 +360,11 @@ func (c *SourceConfig) AdjustServerID(ctx context.Context, db *sql.DB) error {
 
 // LoadFromFile loads config from file.
 func LoadFromFile(path string) (*SourceConfig, error) {
-	c := newSourceConfig()
 	content, err := os.ReadFile(path)
 	if err != nil {
 		return nil, terror.ErrConfigReadCfgFromFile.Delegate(err, path)
 	}
-	if err = yaml.UnmarshalStrict(content, c); err != nil {
-		return nil, terror.ErrConfigYamlTransform.Delegate(err, "decode source config")
-	}
-	c.adjust()
-	if err = c.Verify(); err != nil {
-		return nil, err
-	}
-	return c, nil
+	return ParseYaml(string(content))
 }
 
 func (c *SourceConfig) check(metaData *toml.MetaData, err error) error {

dm/dm/config/source_config_test.go

@@ -31,11 +31,8 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 )
 
-// do not forget to update this path if the file removed/renamed.
-const sourceSampleFile = "../worker/source.yaml"
-
 func (t *testConfig) TestConfig(c *C) {
-	cfg, err := LoadFromFile(sourceSampleFile)
+	cfg, err := ParseYaml(SampleSourceConfig)
 	c.Assert(err, IsNil)
 	cfg.RelayDir = "./xx"
 	c.Assert(cfg.RelayDir, Equals, "./xx")
@@ -134,7 +131,7 @@ aaa: xxx
 
 func (t *testConfig) TestConfigVerify(c *C) {
 	newConfig := func() *SourceConfig {
-		cfg, err := LoadFromFile(sourceSampleFile)
+		cfg, err := ParseYaml(SampleSourceConfig)
 		c.Assert(err, IsNil)
 		cfg.RelayDir = "./xx"
 		return cfg
@@ -240,7 +237,7 @@ func (t *testConfig) TestConfigVerify(c *C) {
 }
 
 func (t *testConfig) TestSourceConfigForDowngrade(c *C) {
-	cfg, err := LoadFromFile(sourceSampleFile)
+	cfg, err := ParseYaml(SampleSourceConfig)
 	c.Assert(err, IsNil)
 
 	// make sure all new field were added
@@ -274,7 +271,7 @@ func subtestFlavor(c *C, cfg *SourceConfig, sqlInfo, expectedFlavor, expectedErr
 }
 
 func (t *testConfig) TestAdjustFlavor(c *C) {
-	cfg, err := LoadFromFile(sourceSampleFile)
+	cfg, err := ParseYaml(SampleSourceConfig)
 	c.Assert(err, IsNil)
 	cfg.RelayDir = "./xx"
 
@@ -297,7 +294,7 @@ func (t *testConfig) TestAdjustServerID(c *C) {
 	}()
 	getAllServerIDFunc = getMockServerIDs
 
-	cfg, err := LoadFromFile(sourceSampleFile)
+	cfg, err := ParseYaml(SampleSourceConfig)
 	c.Assert(err, IsNil)
 	cfg.RelayDir = "./xx"
 
@@ -317,7 +314,7 @@ func getMockServerIDs(ctx context.Context, db *sql.DB) (map[uint32]struct{}, err
 }
 
 func (t *testConfig) TestAdjustCaseSensitive(c *C) {
-	cfg, err := LoadFromFile(sourceSampleFile)
+	cfg, err := ParseYaml(SampleSourceConfig)
 	c.Assert(err, IsNil)
 
 	db, mock, err := sqlmock.New()
@@ -339,5 +336,5 @@ func (t *testConfig) TestAdjustCaseSensitive(c *C) {
 func (t *testConfig) TestEmbedSampleFile(c *C) {
 	data, err := os.ReadFile("./source.yaml")
 	c.Assert(err, IsNil)
-	c.Assert(SampleConfigFile, Equals, string(data))
+	c.Assert(SampleSourceConfig, Equals, string(data))
 }

dm/dm/config/source_converter_test.go

@@ -15,11 +15,12 @@ package config
 
 import (
 	"github.com/pingcap/check"
+
 	"github.com/pingcap/tiflow/dm/openapi/fixtures"
 )
 
 func (t *testConfig) TestConverterWithSourceAndOpenAPISource(c *check.C) {
-	sourceCfg1, err := LoadFromFile(sourceSampleFile)
+	sourceCfg1, err := ParseYaml(SampleSourceConfig)
 	c.Assert(err, check.IsNil)
 
 	// 1. test user create source from dmctl, after convert to openapi.Source then convert back to source config
@@ -28,6 +29,8 @@ func (t *testConfig) TestConverterWithSourceAndOpenAPISource(c *check.C) {
 	// we need set ServerID and MaxAllowedPacket manually, because user don't need to config those field in openapi
 	sourceCfg2.ServerID = sourceCfg1.ServerID
 	sourceCfg2.From.MaxAllowedPacket = sourceCfg1.From.MaxAllowedPacket
+	// TODO: OpenAPI source config will add flavor
+	sourceCfg1.Flavor = ""
 
 	// we only need to make sure the source config that user can see is the same as the source config that user create
 	c.Assert(sourceCfg1.String(), check.Equals, sourceCfg2.String())

dm/dm/config/subtask.go

@@ -15,6 +15,7 @@ package config
 
 import (
 	"bytes"
+	_ "embed"
 	"encoding/json"
 	"flag"
 	"fmt"
@@ -270,6 +271,10 @@ type SubTaskConfig struct {
 	} `yaml:"experimental" toml:"experimental" json:"experimental"`
 }
 
+// SampleSubtaskConfig is the content of subtask.toml in current folder.
+//go:embed subtask.toml
+var SampleSubtaskConfig string
+
 // NewSubTaskConfig creates a new SubTaskConfig.
 func NewSubTaskConfig() *SubTaskConfig {
 	cfg := &SubTaskConfig{}