add message struct (#803)

* add message struct

* use message provided method

* codec use local message

* adjust open protocol

* add some new

* fix build

* add more copyright headers

* fix make check

* remove useless field from the message

* run make check

* add copy right

api/http.go

@@ -29,7 +29,6 @@ func RegisterRoutes(
 	server server.Server,
 	registry prometheus.Gatherer,
 ) {
-
 	// Open API V2
 	v2.RegisterOpenAPIV2Routes(router, v2.NewOpenAPIV2(server))
 

api/middleware/middleware_test.go

@@ -1,3 +1,16 @@
+// Copyright 2025 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package middleware
 
 import (

cmd/cli/cli_changefeed_move_table.go

@@ -1,3 +1,16 @@
+// Copyright 2025 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package cli
 
 import (

cmd/main.go

@@ -49,7 +49,6 @@ func addNewArchCommandTo(cmd *cobra.Command) {
 }
 
 func isNewArchEnabledByConfig(serverConfigFilePath string) bool {
-
 	cfg := config.GetDefaultServerConfig()
 	if len(serverConfigFilePath) > 0 {
 		// strict decode config file, but ignore debug item

cmd/server/server.go

@@ -18,12 +18,11 @@ import (
 	"os"
 	"strings"
 
+	"github.com/fatih/color"
+	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/logger"
-
-	"github.com/fatih/color"
-	"github.com/pingcap/errors"
 	"github.com/pingcap/ticdc/server"
 	"github.com/pingcap/ticdc/version"
 	"github.com/pingcap/tiflow/pkg/cmd/util"

coordinator/changefeed/backoff.go

@@ -167,7 +167,7 @@ func (m *Backoff) HandleError(errs []*heartbeatpb.RunningError) (bool, *heartbea
 		}
 	}
 
-	var lastError = errs[len(errs)-1]
+	lastError := errs[len(errs)-1]
 
 	if !m.retrying.Load() {
 		// errBackoff may be stopped, reset it before the first retry.

coordinator/changefeed/backoff_test.go

@@ -37,7 +37,8 @@ func TestRetry(t *testing.T) {
 	changefeed, state, err := backoff.CheckStatus(&heartbeatpb.MaintainerStatus{
 		CheckpointTs: 1,
 		Err: []*heartbeatpb.RunningError{
-			{Message: "test"}},
+			{Message: "test"},
+		},
 	})
 	require.True(t, changefeed)
 	require.Equal(t, model.StateWarning, state)
@@ -56,12 +57,13 @@ func TestRetry(t *testing.T) {
 	require.True(t, backoff.ShouldRun())
 	require.False(t, backoff.retrying.Load())
 
-	//fail
+	// fail
 	changefeed, state, err = backoff.CheckStatus(&heartbeatpb.MaintainerStatus{
 		CheckpointTs: 1,
 		Err: []*heartbeatpb.RunningError{
 			{Message: "test"},
-			{Code: "CDC:ErrSnapshotLostByGC", Message: "snapshot lost by gc"}},
+			{Code: "CDC:ErrSnapshotLostByGC", Message: "snapshot lost by gc"},
+		},
 	})
 
 	require.True(t, changefeed)
@@ -101,7 +103,8 @@ func TestErrorReportedWhenRetrying(t *testing.T) {
 	changefeed, state, err := backoff.CheckStatus(&heartbeatpb.MaintainerStatus{
 		CheckpointTs: 1,
 		Err: []*heartbeatpb.RunningError{
-			{Message: "test"}},
+			{Message: "test"},
+		},
 	})
 	require.NotNil(t, err)
 	require.True(t, changefeed)
@@ -114,7 +117,8 @@ func TestErrorReportedWhenRetrying(t *testing.T) {
 	changefeed, state, err = backoff.CheckStatus(&heartbeatpb.MaintainerStatus{
 		CheckpointTs: 1,
 		Err: []*heartbeatpb.RunningError{
-			{Message: "test"}},
+			{Message: "test"},
+		},
 	})
 	require.NotNil(t, err)
 	require.True(t, changefeed)
@@ -136,7 +140,8 @@ func TestFailedWhenRetry(t *testing.T) {
 	changefeed, state, err := backoff.CheckStatus(&heartbeatpb.MaintainerStatus{
 		CheckpointTs: 1,
 		Err: []*heartbeatpb.RunningError{
-			{Message: "test"}},
+			{Message: "test"},
+		},
 	})
 	require.NotNil(t, err)
 	require.True(t, changefeed)
@@ -149,7 +154,8 @@ func TestFailedWhenRetry(t *testing.T) {
 	changefeed, state, err = backoff.CheckStatus(&heartbeatpb.MaintainerStatus{
 		CheckpointTs: 1,
 		Err: []*heartbeatpb.RunningError{
-			{Message: "test"}},
+			{Message: "test"},
+		},
 	})
 	require.NotNil(t, err)
 	require.True(t, changefeed)

coordinator/changefeed/changefeed.go

@@ -48,7 +48,8 @@ type Changefeed struct {
 // NewChangefeed creates a new changefeed instance
 func NewChangefeed(cfID common.ChangeFeedID,
 	info *config.ChangeFeedInfo,
-	checkpointTs uint64) *Changefeed {
+	checkpointTs uint64,
+) *Changefeed {
 	uri, err := url.Parse(info.SinkURI)
 	if err != nil {
 		log.Panic("unable to marshal changefeed config",

coordinator/changefeed/changefeed_db_test.go

@@ -210,26 +210,32 @@ func TestReplaceStoppedChangefeed(t *testing.T) {
 	cfID := common.NewChangeFeedIDWithName("test")
 	cf := &Changefeed{
 		ID: cfID,
-		info: atomic.NewPointer(&config.ChangeFeedInfo{ChangefeedID: cfID,
-			Config:  config.GetDefaultReplicaConfig(),
-			SinkURI: "mysql://127.0.0.1:3306"}),
+		info: atomic.NewPointer(&config.ChangeFeedInfo{
+			ChangefeedID: cfID,
+			Config:       config.GetDefaultReplicaConfig(),
+			SinkURI:      "mysql://127.0.0.1:3306",
+		}),
 		status: atomic.NewPointer(&heartbeatpb.MaintainerStatus{}),
 	}
 	db.AddStoppedChangefeed(cf)
 	require.Contains(t, db.stopped, cf.ID)
 
-	cf2 := &config.ChangeFeedInfo{ChangefeedID: cfID,
-		Config:  config.GetDefaultReplicaConfig(),
-		SinkURI: "kafka://127.0.0.1:9092"}
+	cf2 := &config.ChangeFeedInfo{
+		ChangefeedID: cfID,
+		Config:       config.GetDefaultReplicaConfig(),
+		SinkURI:      "kafka://127.0.0.1:9092",
+	}
 	db.ReplaceStoppedChangefeed(cf2)
 	require.Contains(t, db.stopped, cf.ID)
 
 	cf3 := db.GetByID(cf.ID)
 	require.Equal(t, true, cf3.isMQSink)
 
 	cf4ID := common.NewChangeFeedIDWithName("test4")
-	cf4 := &config.ChangeFeedInfo{ChangefeedID: cf4ID,
-		SinkURI: "kafka://127.0.0.1:9092"}
+	cf4 := &config.ChangeFeedInfo{
+		ChangefeedID: cf4ID,
+		SinkURI:      "kafka://127.0.0.1:9092",
+	}
 	db.ReplaceStoppedChangefeed(cf4)
 
 	require.NotContains(t, db.changefeeds, cf4ID)
@@ -238,9 +244,11 @@ func TestReplaceStoppedChangefeed(t *testing.T) {
 func TestScheduleChangefeed(t *testing.T) {
 	db := NewChangefeedDB(1216)
 	cfID := common.NewChangeFeedIDWithName("test")
-	cf := NewChangefeed(cfID, &config.ChangeFeedInfo{ChangefeedID: cfID,
-		Config:  config.GetDefaultReplicaConfig(),
-		SinkURI: "mysql://127.0.0.1:3306"},
+	cf := NewChangefeed(cfID, &config.ChangeFeedInfo{
+		ChangefeedID: cfID,
+		Config:       config.GetDefaultReplicaConfig(),
+		SinkURI:      "mysql://127.0.0.1:3306",
+	},
 		10)
 	db.AddAbsentChangefeed(cf)
 	db.MarkMaintainerScheduling(cf)
@@ -267,36 +275,40 @@ func TestCalculateGCSafepoint(t *testing.T) {
 
 	cfID := common.NewChangeFeedIDWithName("test")
 	cf1 := NewChangefeed(cfID,
-		&config.ChangeFeedInfo{ChangefeedID: cfID,
-			Config: config.GetDefaultReplicaConfig(),
-			State:  model.StateStopped,
+		&config.ChangeFeedInfo{
+			ChangefeedID: cfID,
+			Config:       config.GetDefaultReplicaConfig(),
+			State:        model.StateStopped,
 		}, 11)
 	db.AddStoppedChangefeed(cf1)
 	require.Equal(t, uint64(11), db.CalculateGCSafepoint())
 
 	cf2ID := common.NewChangeFeedIDWithName("test")
 	cf2 := NewChangefeed(cf2ID,
-		&config.ChangeFeedInfo{ChangefeedID: cf2ID,
-			Config: config.GetDefaultReplicaConfig(),
-			State:  model.StateFinished,
+		&config.ChangeFeedInfo{
+			ChangefeedID: cf2ID,
+			Config:       config.GetDefaultReplicaConfig(),
+			State:        model.StateFinished,
 		}, 9)
 	db.AddStoppedChangefeed(cf2)
 	require.Equal(t, uint64(11), db.CalculateGCSafepoint())
 
 	cf3ID := common.NewChangeFeedIDWithName("test")
 	cf3 := NewChangefeed(cf3ID,
-		&config.ChangeFeedInfo{ChangefeedID: cf3ID,
-			Config: config.GetDefaultReplicaConfig(),
-			State:  model.StateNormal,
+		&config.ChangeFeedInfo{
+			ChangefeedID: cf3ID,
+			Config:       config.GetDefaultReplicaConfig(),
+			State:        model.StateNormal,
 		}, 10)
 	db.AddStoppedChangefeed(cf3)
 	require.Equal(t, uint64(10), db.CalculateGCSafepoint())
 
 	cf4ID := common.NewChangeFeedIDWithName("test")
 	cf4 := NewChangefeed(cf4ID,
-		&config.ChangeFeedInfo{ChangefeedID: cf4ID,
-			Config: config.GetDefaultReplicaConfig(),
-			State:  model.StateFailed,
+		&config.ChangeFeedInfo{
+			ChangefeedID: cf4ID,
+			Config:       config.GetDefaultReplicaConfig(),
+			State:        model.StateFailed,
 			Error: &model.RunningError{
 				Code: string(errors.ErrGCTTLExceeded.ID()),
 			},
@@ -306,10 +318,11 @@ func TestCalculateGCSafepoint(t *testing.T) {
 
 	cf5ID := common.NewChangeFeedIDWithName("test")
 	cf5 := NewChangefeed(cf5ID,
-		&config.ChangeFeedInfo{ChangefeedID: cf5ID,
-			Config: config.GetDefaultReplicaConfig(),
-			State:  model.StateFailed,
-			Error:  &model.RunningError{},
+		&config.ChangeFeedInfo{
+			ChangefeedID: cf5ID,
+			Config:       config.GetDefaultReplicaConfig(),
+			State:        model.StateFailed,
+			Error:        &model.RunningError{},
 		}, 7)
 	db.AddStoppedChangefeed(cf5)
 	require.Equal(t, uint64(7), db.CalculateGCSafepoint())

coordinator/changefeed/etcd_backend.go

@@ -125,7 +125,8 @@ func (b *EtcdBackend) GetAllChangefeeds(ctx context.Context) (map[common.ChangeF
 }
 
 func (b *EtcdBackend) CreateChangefeed(ctx context.Context,
-	info *config.ChangeFeedInfo) error {
+	info *config.ChangeFeedInfo,
+) error {
 	infoKey := etcd.GetEtcdKeyChangeFeedInfo(b.etcdClient.GetClusterID(), info.ChangefeedID.DisplayName)
 	infoValue, err := info.Marshal()
 	if err != nil {
@@ -229,7 +230,8 @@ func (b *EtcdBackend) PauseChangefeed(ctx context.Context, id common.ChangeFeedI
 }
 
 func (b *EtcdBackend) DeleteChangefeed(ctx context.Context,
-	changefeedID common.ChangeFeedID) error {
+	changefeedID common.ChangeFeedID,
+) error {
 	infoKey := etcd.GetEtcdKeyChangeFeedInfo(b.etcdClient.GetClusterID(), changefeedID.DisplayName)
 	jobKey := etcd.GetEtcdKeyJob(b.etcdClient.GetClusterID(), changefeedID.DisplayName)
 	opsThen := []clientv3.Op{}
@@ -247,7 +249,8 @@ func (b *EtcdBackend) DeleteChangefeed(ctx context.Context,
 }
 
 func (b *EtcdBackend) ResumeChangefeed(ctx context.Context,
-	id common.ChangeFeedID, newCheckpointTs uint64) error {
+	id common.ChangeFeedID, newCheckpointTs uint64,
+) error {
 	info, err := b.etcdClient.GetChangeFeedInfo(ctx, id.DisplayName)
 	if err != nil {
 		return errors.Trace(err)

coordinator/controller.go

@@ -87,7 +87,8 @@ func NewController(
 	backend changefeed.Backend,
 	stream dynstream.DynamicStream[int, string, *Event, *Controller, *StreamHandler],
 	taskScheduler threadpool.ThreadPool,
-	batchSize int, balanceInterval time.Duration) *Controller {
+	batchSize int, balanceInterval time.Duration,
+) *Controller {
 	mc := appcontext.GetService[messaging.MessageCenter](appcontext.MessageCenter)
 	changefeedDB := changefeed.NewChangefeedDB(version)
 
@@ -123,7 +124,7 @@ func NewController(
 	})
 	log.Info("changefeed bootstrap initial nodes",
 		zap.Int("nodes", len(nodes)))
-	var newNodes = make([]*node.Info, 0, len(nodes))
+	newNodes := make([]*node.Info, 0, len(nodes))
 	for _, n := range nodes {
 		newNodes = append(newNodes, n)
 	}
@@ -191,14 +192,14 @@ func (c *Controller) onNodeChanged() {
 	currentNodes := c.bootstrapper.GetAllNodes()
 
 	activeNodes := c.nodeManager.GetAliveNodes()
-	var newNodes = make([]*node.Info, 0, len(activeNodes))
+	newNodes := make([]*node.Info, 0, len(activeNodes))
 	for id, n := range activeNodes {
 		if _, ok := currentNodes[id]; !ok {
 			newNodes = append(newNodes, n)
 		}
 	}
 	var removedNodes []node.ID
-	for id, _ := range currentNodes {
+	for id := range currentNodes {
 		if _, ok := activeNodes[id]; !ok {
 			removedNodes = append(removedNodes, id)
 			c.RemoveNode(id)