br: check the correct changefeed info when restore/import data

br/pkg/task/stream.go

@@ -1076,12 +1076,14 @@ func checkTaskExists(ctx context.Context, cfg *RestoreConfig, etcdCLI *clientv3.
 	}
 
 	// check cdc changefeed
-	nameSet, err := utils.GetCDCChangefeedNameSet(ctx, etcdCLI)
-	if err != nil {
-		return err
-	}
-	if !nameSet.Empty() {
-		return errors.Errorf("%splease stop changefeed(s) before restore", nameSet.MessageToUser())
+	if cfg.CheckRequirements {
+		nameSet, err := utils.GetCDCChangefeedNameSet(ctx, etcdCLI)
+		if err != nil {
+			return err
+		}
+		if !nameSet.Empty() {
+			return errors.Errorf("%splease stop changefeed(s) before restore", nameSet.MessageToUser())
+		}
 	}
 	return nil
 }

br/pkg/utils/cdc.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"regexp"
 	"strings"
 
 	"github.com/pingcap/errors"
@@ -68,17 +69,42 @@ func GetCDCChangefeedNameSet(ctx context.Context, cli *clientv3.Client) (*CDCNam
 		return nil, errors.Trace(err)
 	}
 
+	// Generate a random cluster ID in pd
+	// r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	// ts := uint64(time.Now().Unix())
+	// clusterID := (ts << 32) + uint64(r.Uint32())
+	reg, err := regexp.Compile("^[a-zA-Z0-9]+(-[a-zA-Z0-9]+)*$")
+	if err != nil {
+		log.L().Warn("failed to parse cluster id, skip it", zap.Error(err))
+		reg = nil
+	}
+
 	for _, kv := range resp.Kvs {
 		// example: /tidb/cdc/<clusterID>/<namespace>/changefeed/info/<changefeedID>
 		k := kv.Key[len(CDCPrefix):]
 		clusterAndNamespace, changefeedID, found := bytes.Cut(k, []byte(ChangefeedPath))
 		if !found {
 			continue
 		}
-		if !isActiveCDCChangefeed(kv.Value) {
+		// example: clusterAndNamespace normally is <clusterID>/<namespace>
+		// but in migration scenario it become __backup__. we need handle it
+		// see https://github.com/pingcap/tiflow/issues/9807
+		clusterID, _, found := bytes.Cut(clusterAndNamespace, []byte(`/`))
+		if !found {
+			// ignore __backup__ or other formats
 			continue
 		}
 
+		if reg != nil {
+			matched := reg.Match(clusterID)
+			if !matched {
+				continue
+			}
+			if !isActiveCDCChangefeed(kv.Value) {
+				continue
+			}
+		}
+
 		nameSet[string(clusterAndNamespace)] = append(nameSet[string(clusterAndNamespace)], string(changefeedID))
 	}
 	if len(nameSet) == 0 {

br/pkg/utils/cdc_test.go

@@ -85,4 +85,21 @@ func TestGetCDCChangefeedNameSet(t *testing.T) {
 	require.False(t, nameSet.Empty())
 	require.Equal(t, "found CDC changefeed(s): cluster/namespace: <nil> changefeed(s): [test], ",
 		nameSet.MessageToUser())
+
+	_, err = cli.Delete(ctx, "/tidb/cdc/", clientv3.WithPrefix())
+	require.NoError(t, err)
+
+	// ignore __backup__ changefeed
+	checkEtcdPut(
+		"/tidb/cdc/__backup__/changefeed/info/test",
+		`{"upstream-id":7195826648407968958,"namespace":"default","changefeed-id":"test-1","sink-uri":"mysql://root@127.0.0.1:3306?time-zone=","create-time":"2023-02-03T15:23:34.773768+08:00","start-ts":439198420741652483,"target-ts":0,"admin-job-type":0,"sort-engine":"unified","sort-dir":"","config":{"memory-quota":1073741824,"case-sensitive":true,"enable-old-value":true,"force-replicate":false,"check-gc-safe-point":true,"enable-sync-point":false,"bdr-mode":false,"sync-point-interval":600000000000,"sync-point-retention":86400000000000,"filter":{"rules":["*.*"],"ignore-txn-start-ts":null,"event-filters":null},"mounter":{"worker-num":16},"sink":{"transaction-atomicity":"","protocol":"","dispatchers":null,"csv":{"delimiter":",","quote":"\"","null":"\\N","include-commit-ts":false},"column-selectors":null,"schema-registry":"","encoder-concurrency":16,"terminator":"\r\n","date-separator":"none","enable-partition-separator":false},"consistent":{"level":"none","max-log-size":64,"flush-interval":2000,"storage":""},"scheduler":{"region-per-span":0}},"state":"normal","error":null,"creator-version":"v6.5.0-master-dirty"}`,
+	)
+	// ignore cluster id only changefeed
+	checkEtcdPut(
+		"/tidb/cdc/5402613591834624000/changefeed/info/test",
+		`{"upstream-id":7195826648407968958,"namespace":"default","changefeed-id":"test-1","sink-uri":"mysql://root@127.0.0.1:3306?time-zone=","create-time":"2023-02-03T15:23:34.773768+08:00","start-ts":439198420741652483,"target-ts":0,"admin-job-type":0,"sort-engine":"unified","sort-dir":"","config":{"memory-quota":1073741824,"case-sensitive":true,"enable-old-value":true,"force-replicate":false,"check-gc-safe-point":true,"enable-sync-point":false,"bdr-mode":false,"sync-point-interval":600000000000,"sync-point-retention":86400000000000,"filter":{"rules":["*.*"],"ignore-txn-start-ts":null,"event-filters":null},"mounter":{"worker-num":16},"sink":{"transaction-atomicity":"","protocol":"","dispatchers":null,"csv":{"delimiter":",","quote":"\"","null":"\\N","include-commit-ts":false},"column-selectors":null,"schema-registry":"","encoder-concurrency":16,"terminator":"\r\n","date-separator":"none","enable-partition-separator":false},"consistent":{"level":"none","max-log-size":64,"flush-interval":2000,"storage":""},"scheduler":{"region-per-span":0}},"state":"normal","error":null,"creator-version":"v6.5.0-master-dirty"}`,
+	)
+	nameSet, err = utils.GetCDCChangefeedNameSet(ctx, cli)
+	require.NoError(t, err)
+	require.True(t, nameSet.Empty())
 }