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

…pingcap#10452)

close pingcap#10430

cdc/api/v1/api.go

@@ -307,9 +307,7 @@ func (h *OpenAPI) CreateChangefeed(c *gin.Context) {
 		CAPath:        up.SecurityConfig.CAPath,
 		CertAllowedCN: up.SecurityConfig.CertAllowedCN,
 	}
-	err = h.capture.GetEtcdClient().CreateChangefeedInfo(
-		ctx, upstreamInfo,
-		info, model.DefaultChangeFeedID(changefeedConfig.ID))
+	err = h.capture.GetEtcdClient().CreateChangefeedInfo(ctx, upstreamInfo, info)
 	if err != nil {
 		_ = c.Error(err)
 		return

cdc/api/v2/changefeed.go

@@ -154,10 +154,7 @@ func (h *OpenAPIV2) createChangefeed(c *gin.Context) {
 		return
 	}
 
-	err = h.capture.GetEtcdClient().CreateChangefeedInfo(ctx,
-		upstreamInfo,
-		info,
-		model.DefaultChangeFeedID(info.ID))
+	err = h.capture.GetEtcdClient().CreateChangefeedInfo(ctx, upstreamInfo, info)
 	if err != nil {
 		needRemoveGCSafePoint = true
 		_ = c.Error(err)
@@ -453,7 +450,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

@@ -464,7 +464,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()
@@ -484,7 +484,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()
@@ -501,7 +501,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

@@ -1241,6 +1241,11 @@ error = '''
 meta operation fail
 '''
 
+["DFLOW:ErrMetaOpFailed"]
+error = '''
+unexpected meta operation failure: %s
+'''
+
 ["DFLOW:ErrMetaOptionConflict"]
 error = '''
 WithRange/WithPrefix/WithFromKey, more than one option are used

pkg/errors/cdc_errors.go

@@ -880,4 +880,9 @@ var (
 			"add `ignore-txn-start-ts=[%d]` to the changefeed in the filter configuration.",
 		errors.RFCCodeText("CDC:ErrHandleDDLFailed"),
 	)
+
+	ErrMetaOpFailed = errors.Normalize(
+		"unexpected meta operation failure: %s",
+		errors.RFCCodeText("DFLOW:ErrMetaOpFailed"),
+	)
 )

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)