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

close pingcap#10430

cdc/api/v2/changefeed.go

@@ -493,7 +493,7 @@ func (h *OpenAPIV2) updateChangefeed(c *gin.Context) {
 		zap.Any("upstreamInfo", newUpInfo))
 
 	err = h.capture.GetEtcdClient().
-		UpdateChangefeedAndUpstream(ctx, newUpInfo, newCfInfo, changefeedID)
+		UpdateChangefeedAndUpstream(ctx, newUpInfo, newCfInfo)
 	if err != nil {
 		_ = c.Error(errors.Trace(err))
 		return

cdc/api/v2/changefeed_test.go

@@ -475,7 +475,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		Return(&model.ChangeFeedInfo{}, &model.UpstreamInfo{}, nil).
 		Times(1)
 	etcdClient.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()
@@ -495,7 +495,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		Times(1)
 	mockCapture.EXPECT().GetUpstreamManager().Return(upstream.NewManager4Test(&mockPDClient{}), nil).AnyTimes()
 	etcdClient.EXPECT().
-		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil).Times(1)
 
 	w = httptest.NewRecorder()
@@ -512,7 +512,7 @@ func TestUpdateChangefeed(t *testing.T) {
 		Times(1)
 	mockCapture.EXPECT().GetUpstreamManager().Return(upstream.NewManager4Test(&mockPDClient{}), nil).AnyTimes()
 	etcdClient.EXPECT().
-		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+		UpdateChangefeedAndUpstream(gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil).Times(1)
 
 	w = httptest.NewRecorder()

errors.toml

@@ -1343,7 +1343,7 @@ meta operation fail
 
 ["DFLOW:ErrMetaOpFailed"]
 error = '''
-meta operation %s is failed
+unexpected meta operation failure: %s
 '''
 
 ["DFLOW:ErrMetaOptionConflict"]

pkg/errors/cdc_errors.go

@@ -958,7 +958,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(

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)