Modernize cdc

cdc/api/middleware/middleware.go

@@ -15,6 +15,7 @@ package middleware
 
 import (
 	"net/http"
+	"slices"
 	"time"
 
 	"github.com/gin-gonic/gin"
@@ -154,11 +155,8 @@ func verify(ctx *gin.Context, up *upstream.Upstream) error {
 
 	allowed := false
 	serverCfg := config.GetGlobalServerConfig()
-	for _, user := range serverCfg.Security.ClientAllowedUser {
-		if user == username {
-			allowed = true
-			break
-		}
+	if slices.Contains(serverCfg.Security.ClientAllowedUser, username) {
+		allowed = true
 	}
 	if !allowed {
 		errMsg := "The user is not allowed."

cdc/api/util.go

@@ -104,7 +104,7 @@ func WriteError(w http.ResponseWriter, statusCode int, err error) {
 }
 
 // WriteData write data to response with http status code 200
-func WriteData(w http.ResponseWriter, data interface{}) {
+func WriteData(w http.ResponseWriter, data any) {
 	js, err := json.MarshalIndent(data, "", " ")
 	if err != nil {
 		log.Error("invalid json data", zap.Any("data", data), zap.Error(err))

cdc/api/v2/model.go

@@ -162,7 +162,7 @@ func (d JSONDuration) MarshalJSON() ([]byte, error) {
 
 // UnmarshalJSON unmarshal json value to wrapped duration
 func (d *JSONDuration) UnmarshalJSON(b []byte) error {
-	var v interface{}
+	var v any
 	if err := json.Unmarshal(b, &v); err != nil {
 		return err
 	}

cdc/api/v2/unsafe_test.go

@@ -80,8 +80,7 @@ func TestCDCMetaData(t *testing.T) {
 }
 
 func TestWithUpstreamConfig(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	ctx := t.Context()
 	upManager := upstream.NewManager(ctx, upstream.CaptureTopologyCfg{GCServiceID: "abc"})
 	upManager.AddUpstream(&model.UpstreamInfo{
 		ID:          uint64(1),

cdc/capture/capture_test.go

@@ -64,12 +64,10 @@ func TestReset(t *testing.T) {
 	// simulate network isolation scenarios
 	etcdServer.Close()
 	wg := sync.WaitGroup{}
-	wg.Add(1)
-	go func() {
+	wg.Go(func() {
 		_, err = cp.reset(ctx)
 		require.Regexp(t, ".*context canceled.*", err)
-		wg.Done()
-	}()
+	})
 	time.Sleep(100 * time.Millisecond)
 	info, err := cp.Info()
 	require.Nil(t, err)
@@ -230,15 +228,13 @@ func TestCampaignLiveness(t *testing.T) {
 	// Force set alive.
 	cp.liveness = model.LivenessCaptureAlive
 	wg := sync.WaitGroup{}
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
+	wg.Go(func() {
 		// Grant campaign
 		g := <-me.campaignRequestCh
 		// Set liveness to stopping
 		cp.liveness.Store(model.LivenessCaptureStopping)
 		me.campaignGrantCh <- g
-	}()
+	})
 	err = cp.campaignOwner(ctx, globalVars)
 	require.Nil(t, err)
 	require.True(t, me.campaignFlag)

cdc/entry/mounter.go

@@ -355,7 +355,7 @@ func datum2Column(
 		colDatum, exist := datums[colID]
 
 		var (
-			colValue interface{}
+			colValue any
 			size     int
 			warn     string
 			err      error
@@ -459,7 +459,7 @@ func (m *mounter) verifyColumnChecksum(
 	return checksum, true, nil
 }
 
-func newDatum(value interface{}, ft types.FieldType) (types.Datum, error) {
+func newDatum(value any, ft types.FieldType) (types.Datum, error) {
 	if value == nil {
 		return types.NewDatum(nil), nil
 	}
@@ -770,7 +770,7 @@ func sizeOfBytes(b []byte) int {
 
 // formatColVal return interface{} need to meet the same requirement as getDefaultOrZeroValue
 func formatColVal(datum types.Datum, col *timodel.ColumnInfo) (
-	value interface{}, size int, warn string, err error,
+	value any, size int, warn string, err error,
 ) {
 	if datum.IsNull() {
 		return nil, 0, "", nil

cdc/entry/mounter_test.go

@@ -59,33 +59,33 @@ func TestMounterDisableOldValue(t *testing.T) {
 		tableName      string
 		createTableDDL string
 		// [] for rows, []interface{} for columns.
-		values [][]interface{}
+		values [][]any
 		// [] for table partition if there is any,
 		// []int for approximateBytes of rows.
 		putApproximateBytes [][]int
 		delApproximateBytes [][]int
 	}{{
 		tableName:           "simple",
 		createTableDDL:      "create table simple(id int primary key)",
-		values:              [][]interface{}{{1}, {2}, {3}, {4}, {5}},
+		values:              [][]any{{1}, {2}, {3}, {4}, {5}},
 		putApproximateBytes: [][]int{{346, 346, 346, 346, 346}},
 		delApproximateBytes: [][]int{{346, 346, 346, 346, 346}},
 	}, {
 		tableName:           "no_pk",
 		createTableDDL:      "create table no_pk(id int not null unique key)",
-		values:              [][]interface{}{{1}, {2}, {3}, {4}, {5}},
+		values:              [][]any{{1}, {2}, {3}, {4}, {5}},
 		putApproximateBytes: [][]int{{345, 345, 345, 345, 345}},
 		delApproximateBytes: [][]int{{217, 217, 217, 217, 217}},
 	}, {
 		tableName:           "many_index",
 		createTableDDL:      "create table many_index(id int not null unique key, c1 int unique key, c2 int, INDEX (c2))",
-		values:              [][]interface{}{{1, 1, 1}, {2, 2, 2}, {3, 3, 3}, {4, 4, 4}, {5, 5, 5}},
+		values:              [][]any{{1, 1, 1}, {2, 2, 2}, {3, 3, 3}, {4, 4, 4}, {5, 5, 5}},
 		putApproximateBytes: [][]int{{638, 638, 638, 638, 638}},
 		delApproximateBytes: [][]int{{254, 254, 254, 254, 254}},
 	}, {
 		tableName:           "default_value",
 		createTableDDL:      "create table default_value(id int primary key, c1 int, c2 int not null default 5, c3 varchar(20), c4 varchar(20) not null default '666')",
-		values:              [][]interface{}{{1}, {2}, {3}, {4}, {5}},
+		values:              [][]any{{1}, {2}, {3}, {4}, {5}},
 		putApproximateBytes: [][]int{{676, 676, 676, 676, 676}},
 		delApproximateBytes: [][]int{{353, 353, 353, 353, 353}},
 	}, {
@@ -105,7 +105,7 @@ func TestMounterDisableOldValue(t *testing.T) {
 				PARTITION p2 VALUES LESS THAN (15),
 				PARTITION p3 VALUES LESS THAN (20)
 			)`,
-		values: [][]interface{}{
+		values: [][]any{
 			{1, "aa", "bb", 12, 12},
 			{6, "aac", "bab", 51, 51},
 			{11, "aad", "bsb", 71, 61},
@@ -127,7 +127,7 @@ func TestMounterDisableOldValue(t *testing.T) {
 				constraint pk
 					primary key (id)
 			);`,
-		values: [][]interface{}{
+		values: [][]any{
 			{1, 1, 2, 3, 4, 5},
 			{2},
 			{3, 3, 4, 5, 6, 7},
@@ -156,7 +156,7 @@ func TestMounterDisableOldValue(t *testing.T) {
 			constraint pk
 			primary key (id)
 		);`,
-		values: [][]interface{}{
+		values: [][]any{
 			{1},
 			{
 				2, "89504E470D0A1A0A", "89504E470D0A1A0A", "89504E470D0A1A0A", "89504E470D0A1A0A", "89504E470D0A1A0A",
@@ -191,7 +191,7 @@ func TestMounterDisableOldValue(t *testing.T) {
 			constraint pk
 			primary key (id)
 		);`,
-		values: [][]interface{}{
+		values: [][]any{
 			{1},
 			{2, "2020-02-20", "2020-02-20 02:20:20", "2020-02-20 02:20:20", "02:20:20", "2020"},
 		},
@@ -208,7 +208,7 @@ func TestMounterDisableOldValue(t *testing.T) {
 		constraint pk
 		primary key (id)
 	);`,
-		values: [][]interface{}{
+		values: [][]any{
 			{1},
 			{2, "2020.0202", "2020.0303", "2020.0404"},
 		},
@@ -226,7 +226,7 @@ func TestMounterDisableOldValue(t *testing.T) {
 			constraint pk
 			primary key (id)
 		);`,
-		values: [][]interface{}{
+		values: [][]any{
 			{1},
 			{2, "a", "a,c", 888, `{"aa":"bb"}`},
 		},
@@ -235,7 +235,7 @@ func TestMounterDisableOldValue(t *testing.T) {
 	}, {
 		tableName:      "clustered_index1",
 		createTableDDL: "CREATE TABLE clustered_index1 (id VARCHAR(255) PRIMARY KEY, data INT);",
-		values: [][]interface{}{
+		values: [][]any{
 			{"hhh"},
 			{"你好😘", 666},
 			{"世界🤪", 888},
@@ -245,7 +245,7 @@ func TestMounterDisableOldValue(t *testing.T) {
 	}, {
 		tableName:      "clustered_index2",
 		createTableDDL: "CREATE TABLE clustered_index2 (id VARCHAR(255), data INT, ddaa date, PRIMARY KEY (id, data, ddaa), UNIQUE KEY (id, data, ddaa));",
-		values: [][]interface{}{
+		values: [][]any{
 			{"你好😘", 666, "2020-11-20"},
 			{"世界🤪", 888, "2020-05-12"},
 		},
@@ -260,7 +260,7 @@ func TestMounterDisableOldValue(t *testing.T) {
 func testMounterDisableOldValue(t *testing.T, tc struct {
 	tableName           string
 	createTableDDL      string
-	values              [][]interface{}
+	values              [][]any
 	putApproximateBytes [][]int
 	delApproximateBytes [][]int
 },
@@ -338,12 +338,12 @@ func testMounterDisableOldValue(t *testing.T, tc struct {
 			if len(row.Columns) != 0 {
 				checkSQL, params := prepareCheckSQL(t, tc.tableName, row.GetColumns())
 				result := tk.MustQuery(checkSQL, params...)
-				result.Check([][]interface{}{{"1"}})
+				result.Check([][]any{{"1"}})
 			}
 			if len(row.PreColumns) != 0 {
 				checkSQL, params := prepareCheckSQL(t, tc.tableName, row.GetPreColumns())
 				result := tk.MustQuery(checkSQL, params...)
-				result.Check([][]interface{}{{"1"}})
+				result.Check([][]any{{"1"}})
 			}
 		})
 		return rows
@@ -387,7 +387,7 @@ func prepareInsertSQL(t *testing.T, tableInfo *model.TableInfo, columnLens int)
 	var sb strings.Builder
 	_, err := sb.WriteString("INSERT INTO " + tableInfo.Name.O + "(")
 	require.Nil(t, err)
-	for i := 0; i < columnLens; i++ {
+	for i := range columnLens {
 		col := tableInfo.Columns[i]
 		if i != 0 {
 			_, err = sb.WriteString(", ")
@@ -398,7 +398,7 @@ func prepareInsertSQL(t *testing.T, tableInfo *model.TableInfo, columnLens int)
 	}
 	_, err = sb.WriteString(") VALUES (")
 	require.Nil(t, err)
-	for i := 0; i < columnLens; i++ {
+	for i := range columnLens {
 		if i != 0 {
 			_, err = sb.WriteString(", ")
 			require.Nil(t, err)
@@ -411,11 +411,11 @@ func prepareInsertSQL(t *testing.T, tableInfo *model.TableInfo, columnLens int)
 	return sb.String()
 }
 
-func prepareCheckSQL(t *testing.T, tableName string, cols []*model.Column) (string, []interface{}) {
+func prepareCheckSQL(t *testing.T, tableName string, cols []*model.Column) (string, []any) {
 	var sb strings.Builder
 	_, err := sb.WriteString("SELECT count(1) FROM " + tableName + " WHERE ")
 	require.Nil(t, err)
-	params := make([]interface{}, 0, len(cols))
+	params := make([]any, 0, len(cols))
 	for i, col := range cols {
 		// Since float type has precision problem, so skip it to avoid compare float number.
 		if col == nil || col.Type == mysql.TypeFloat {
@@ -610,7 +610,7 @@ func TestGetDefaultZeroValue(t *testing.T) {
 	testCases := []struct {
 		Name    string
 		ColInfo timodel.ColumnInfo
-		Res     interface{}
+		Res     any
 	}{
 		{
 			Name:    "mysql flag null",
@@ -1352,28 +1352,28 @@ func TestDecodeEventIgnoreRow(t *testing.T) {
 	type testCase struct {
 		schema  string
 		table   string
-		columns []interface{}
+		columns []any
 		ignored bool
 	}
 
 	testCases := []testCase{
 		{
 			schema:  "test",
 			table:   "student",
-			columns: []interface{}{1, "dongmen", 20, "male"},
+			columns: []any{1, "dongmen", 20, "male"},
 			ignored: false,
 		},
 		{
 			schema:  "test",
 			table:   "computer",
-			columns: []interface{}{1, "apple", 19999},
+			columns: []any{1, "apple", 19999},
 			ignored: false,
 		},
 		// This case should be ignored by its table name.
 		{
 			schema:  "test",
 			table:   "poet",
-			columns: []interface{}{1, "李白", "静夜思"},
+			columns: []any{1, "李白", "静夜思"},
 			ignored: true,
 		},
 	}
@@ -1642,14 +1642,14 @@ func TestNewDMRowChange(t *testing.T) {
 		recoveredTI := model.BuildTiDBTableInfo(cdcTableInfo.TableName.Table, cols, cdcTableInfo.IndexColumnsOffset)
 		require.Equal(t, c.recovered, showCreateTable(t, recoveredTI))
 		tableName := &model.TableName{Schema: "db", Table: "t1"}
-		rowChange := sqlmodel.NewRowChange(tableName, nil, []interface{}{1, 1, 2}, nil, recoveredTI, nil, nil)
+		rowChange := sqlmodel.NewRowChange(tableName, nil, []any{1, 1, 2}, nil, recoveredTI, nil, nil)
 		sqlGot, argsGot := rowChange.GenSQL(sqlmodel.DMLDelete)
 		require.Equal(t, "DELETE FROM `db`.`t1` WHERE `a1` = ? AND `a3` = ? LIMIT 1", sqlGot)
-		require.Equal(t, []interface{}{1, 2}, argsGot)
+		require.Equal(t, []any{1, 2}, argsGot)
 
 		sqlGot, argsGot = sqlmodel.GenDeleteSQL(rowChange, rowChange)
 		require.Equal(t, "DELETE FROM `db`.`t1` WHERE (`a1` = ? AND `a3` = ?) OR (`a1` = ? AND `a3` = ?)", sqlGot)
-		require.Equal(t, []interface{}{1, 2, 1, 2}, argsGot)
+		require.Equal(t, []any{1, 2, 1, 2}, argsGot)
 	}
 }
 

cdc/entry/schema/snapshot.go

@@ -1343,7 +1343,7 @@ type versionedID struct {
 	id  int64
 	tag uint64 // A transform of timestamp to reverse sort versions.
 	// the associated entity pointer.
-	target interface{}
+	target any
 }
 
 func versionedEntityNameLess(v1, v2 versionedEntityName) bool {
@@ -1370,18 +1370,18 @@ func newVersionedEntityName(prefix int64, entity string, tag uint64) versionedEn
 // newVersionedID creates an instance with target nil, which means it's deleted from the
 // associated snapshot.
 func newVersionedID(id int64, tag uint64) versionedID {
-	var target interface{} = nil
+	var target any = nil
 	return versionedID{id, tag, target}
 }
 
-func targetToTableInfo(target interface{}) *model.TableInfo {
+func targetToTableInfo(target any) *model.TableInfo {
 	if target == nil {
 		return nil
 	}
 	return target.(*model.TableInfo)
 }
 
-func targetToDBInfo(target interface{}) *timodel.DBInfo {
+func targetToDBInfo(target any) *timodel.DBInfo {
 	if target == nil {
 		return nil
 	}

cdc/entry/schema_storage_test.go

@@ -441,7 +441,7 @@ func TestHandleRenameTables(t *testing.T) {
 	newTableNames := []pmodel.CIStr{pmodel.NewCIStr("x"), pmodel.NewCIStr("y")}
 	oldTableNames := []pmodel.CIStr{pmodel.NewCIStr("oldx"), pmodel.NewCIStr("oldy")}
 	oldSchemaNames := []pmodel.CIStr{pmodel.NewCIStr("db_1"), pmodel.NewCIStr("db_2")}
-	args := []interface{}{oldSchemaIDs, newSchemaIDs, newTableNames, oldTableIDs, oldSchemaNames, oldTableNames}
+	args := []any{oldSchemaIDs, newSchemaIDs, newTableNames, oldTableIDs, oldSchemaNames, oldTableNames}
 	rawArgs, err := json.Marshal(args)
 	require.Nil(t, err)
 	job := &timodel.Job{

cdc/entry/schema_test.go

@@ -16,7 +16,7 @@ package entry
 import (
 	"context"
 	"fmt"
-	"sort"
+	"slices"
 	"testing"
 
 	timodel "github.com/pingcap/tidb/pkg/meta/model"
@@ -78,9 +78,7 @@ func TestAllPhysicalTables(t *testing.T) {
 		expectedTableIDs = append(expectedTableIDs, p.ID)
 	}
 	sortTableIDs := func(tableIDs []model.TableID) {
-		sort.Slice(tableIDs, func(i, j int) bool {
-			return tableIDs[i] < tableIDs[j]
-		})
+		slices.Sort(tableIDs)
 	}
 	sortTableIDs(expectedTableIDs)
 	tableIDs, err = schema.AllPhysicalTables(context.Background(), job.BinlogInfo.FinishedTS)