api(ticdc): only update upstreamInfo that has changed (#10422)

close #10430
pingcap · Jan 11, 2024 · c5d5eff · c5d5eff
1 parent 1bbb996
commit c5d5eff
Show file tree

Hide file tree

Showing 13 changed files with 180 additions and 133 deletions.
diff --git a/cdc/api/v2/changefeed.go b/cdc/api/v2/changefeed.go
@@ -499,7 +499,7 @@ func (h *OpenAPIV2) updateChangefeed(c *gin.Context) {
 		zap.Any("upstreamInfo", newUpInfo))
 
 	err = owner.
-		UpdateChangefeedAndUpstream(ctx, newUpInfo, newCfInfo, changefeedID)
+		UpdateChangefeedAndUpstream(ctx, newUpInfo, newCfInfo)
 	if err != nil {
 		_ = c.Error(errors.Trace(err))
 		return

diff --git a/cdc/api/v2/changefeed_test.go b/cdc/api/v2/changefeed_test.go
@@ -474,7 +474,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		Return(&model.ChangeFeedInfo{}, &model.UpstreamInfo{}, nil).
 		Times(1)
 	mockOwner.EXPECT().
-		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(cerrors.ErrEtcdAPIError).Times(1)
 
 	w = httptest.NewRecorder()
@@ -494,7 +494,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		Times(1)
 	mockCapture.EXPECT().GetUpstreamManager().Return(upstream.NewManager4Test(&mockPDClient{}), nil).AnyTimes()
 	mockOwner.EXPECT().
-		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil).Times(1)
 
 	w = httptest.NewRecorder()
@@ -511,7 +511,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		Times(1)
 	mockCapture.EXPECT().GetUpstreamManager().Return(upstream.NewManager4Test(&mockPDClient{}), nil).AnyTimes()
 	mockOwner.EXPECT().
-		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil).Times(1)
 
 	w = httptest.NewRecorder()

diff --git a/cdc/owner/mock/owner_mock.go b/cdc/owner/mock/owner_mock.go
diff --git a/cdc/owner/owner.go b/cdc/owner/owner.go
@@ -94,7 +94,6 @@ type Owner interface {
 	UpdateChangefeedAndUpstream(ctx context.Context,
 		upstreamInfo *model.UpstreamInfo,
 		changeFeedInfo *model.ChangeFeedInfo,
-		changeFeedID model.ChangeFeedID,
 	) error
 	UpdateChangefeed(ctx context.Context,
 		changeFeedInfo *model.ChangeFeedInfo) error
@@ -405,9 +404,8 @@ func (o *ownerImpl) AsyncStop() {
 func (o *ownerImpl) UpdateChangefeedAndUpstream(ctx context.Context,
 	upstreamInfo *model.UpstreamInfo,
 	changeFeedInfo *model.ChangeFeedInfo,
-	changeFeedID model.ChangeFeedID,
 ) error {
-	return o.etcdClient.UpdateChangefeedAndUpstream(ctx, upstreamInfo, changeFeedInfo, changeFeedID)
+	return o.etcdClient.UpdateChangefeedAndUpstream(ctx, upstreamInfo, changeFeedInfo)
 }
 
 func (o *ownerImpl) UpdateChangefeed(ctx context.Context,

diff --git a/cdcv2/owner/owner.go b/cdcv2/owner/owner.go
@@ -57,7 +57,6 @@ type Owner struct {
 func (o *Owner) UpdateChangefeedAndUpstream(ctx context.Context,
 	upstreamInfo *model.UpstreamInfo,
 	changeFeedInfo *model.ChangeFeedInfo,
-	changeFeedID model.ChangeFeedID,
 ) error {
 	panic("implement me")
 }

diff --git a/errors.toml b/errors.toml
@@ -1353,7 +1353,7 @@ meta operation fail
 
 ["DFLOW:ErrMetaOpFailed"]
 error = '''
-meta operation %s is failed
+unexpected meta operation failure: %s
 '''
 
 ["DFLOW:ErrMetaOptionConflict"]

diff --git a/pkg/errors/cdc_errors.go b/pkg/errors/cdc_errors.go
@@ -962,7 +962,7 @@ var (
 		errors.RFCCodeText("CDC:ErrMetaOpIgnored"),
 	)
 	ErrMetaOpFailed = errors.Normalize(
-		"meta operation %s is failed",
+		"unexpected meta operation failure: %s",
 		errors.RFCCodeText("DFLOW:ErrMetaOpFailed"),
 	)
 	ErrMetaInvalidState = errors.Normalize(

diff --git a/pkg/etcd/client_test.go b/pkg/etcd/client_test.go
@@ -71,6 +71,8 @@ func (m mockWatcher) RequestProgress(ctx context.Context) error {
 }
 
 func TestRetry(t *testing.T) {
+	t.Parallel()
+
 	originValue := maxTries
 	// to speedup the test
 	maxTries = 2
@@ -116,6 +118,8 @@ func TestRetry(t *testing.T) {
 }
 
 func TestDelegateLease(t *testing.T) {
+	t.Parallel()
+
 	ctx := context.Background()
 	url, server, err := SetupEmbedEtcd(t.TempDir())
 	defer func() {
@@ -148,6 +152,8 @@ func TestDelegateLease(t *testing.T) {
 
 // test no data lost when WatchCh blocked
 func TestWatchChBlocked(t *testing.T) {
+	t.Parallel()
+
 	cli := clientv3.NewCtxClient(context.TODO())
 	resetCount := int32(0)
 	requestCount := int32(0)
@@ -209,6 +215,8 @@ func TestWatchChBlocked(t *testing.T) {
 
 // test no data lost when OutCh blocked
 func TestOutChBlocked(t *testing.T) {
+	t.Parallel()
+
 	cli := clientv3.NewCtxClient(context.TODO())
 	resetCount := int32(0)
 	requestCount := int32(0)
@@ -260,8 +268,9 @@ func TestOutChBlocked(t *testing.T) {
 }
 
 func TestRevisionNotFallBack(t *testing.T) {
-	cli := clientv3.NewCtxClient(context.TODO())
+	t.Parallel()
 
+	cli := clientv3.NewCtxClient(context.TODO())
 	resetCount := int32(0)
 	requestCount := int32(0)
 	rev := int64(0)