sink/codc(ticdc): fix schema-registry options in cli & config file (#…

…5469)

ref #5338

cdc/sink/mq/codec/config.go

@@ -108,8 +108,8 @@ func (c *Config) Apply(sinkURI *url.URL, config *config.ReplicaConfig) error {
 		c.avroBigintUnsignedHandlingMode = s
 	}
 
-	if config.SchemaRegistry != "" {
-		c.avroSchemaRegistry = config.SchemaRegistry
+	if config.Sink != nil && config.Sink.SchemaRegistry != "" {
+		c.avroSchemaRegistry = config.Sink.SchemaRegistry
 	}
 
 	return nil

cdc/sink/mq/codec/config_test.go

@@ -52,7 +52,7 @@ func TestConfigApplyValidate(t *testing.T) {
 	c := NewConfig(p)
 	require.Equal(t, config.ProtocolCanalJSON, c.protocol)
 
-	replicaConfig := &config.ReplicaConfig{}
+	replicaConfig := config.GetDefaultReplicaConfig()
 	err = c.Apply(sinkURI, replicaConfig)
 	require.NoError(t, err)
 	require.True(t, c.enableTiDBExtension)
@@ -102,7 +102,7 @@ func TestConfigApplyValidate(t *testing.T) {
 	err = c.Validate()
 	require.ErrorContains(t, err, `Avro protocol requires parameter "schema-registry"`)
 
-	replicaConfig.SchemaRegistry = "this-is-a-uri"
+	replicaConfig.Sink.SchemaRegistry = "this-is-a-uri"
 	err = c.Apply(sinkURI, replicaConfig)
 	require.NoError(t, err)
 	require.Equal(t, "this-is-a-uri", c.avroSchemaRegistry)

docs/swagger/docs.go

@@ -758,6 +758,9 @@ var doc = `{
                 },
                 "protocol": {
                     "type": "string"
+                },
+                "schema-registry": {
+                    "type": "string"
                 }
             }
         },

docs/swagger/swagger.json

@@ -739,6 +739,9 @@
                 },
                 "protocol": {
                     "type": "string"
+                },
+                "schema-registry": {
+                    "type": "string"
                 }
             }
         },

docs/swagger/swagger.yaml

@@ -39,6 +39,8 @@ definitions:
         type: array
       protocol:
         type: string
+      schema-registry:
+        type: string
     type: object
   model.Capture:
     properties:

pkg/cmd/cli/cli_changefeed_create.go

@@ -49,6 +49,7 @@ type changefeedCommonOptions struct {
 	noConfirm              bool
 	targetTs               uint64
 	sinkURI                string
+	schemaRegistry         string
 	configFile             string
 	opts                   []string
 	sortEngine             string
@@ -84,6 +85,8 @@ func (o *changefeedCommonOptions) addFlags(cmd *cobra.Command) {
 	cmd.PersistentFlags().BoolVar(&o.cyclicSyncDDL, "cyclic-sync-ddl", true, "(Experimental) Cyclic replication sync DDL of changefeed")
 	cmd.PersistentFlags().BoolVar(&o.syncPointEnabled, "sync-point", false, "(Experimental) Set and Record syncpoint in replication(default off)")
 	cmd.PersistentFlags().DurationVar(&o.syncPointInterval, "sync-interval", 10*time.Minute, "(Experimental) Set the interval for syncpoint in replication(default 10min)")
+	cmd.PersistentFlags().
+		StringVar(&o.schemaRegistry, "schema-registry", "", "Avro Schema Registry URI")
 	_ = cmd.PersistentFlags().MarkHidden("sort-dir")
 }
 
@@ -113,7 +116,6 @@ type createChangefeedOptions struct {
 	disableGCSafePointCheck bool
 	startTs                 uint64
 	timezone                string
-	schemaRegistry          string
 
 	cfg *config.ReplicaConfig
 }
@@ -137,8 +139,6 @@ func (o *createChangefeedOptions) addFlags(cmd *cobra.Command) {
 	cmd.PersistentFlags().BoolVarP(&o.disableGCSafePointCheck, "disable-gc-check", "", false, "Disable GC safe point check")
 	cmd.PersistentFlags().Uint64Var(&o.startTs, "start-ts", 0, "Start ts of changefeed")
 	cmd.PersistentFlags().StringVar(&o.timezone, "tz", "SYSTEM", "timezone used when checking sink uri (changefeed timezone is determined by cdc server)")
-	cmd.PersistentFlags().
-		StringVar(&o.schemaRegistry, "schema-registry", "", "Avro Schema Registry URI")
 }
 
 // complete adapts from the command line args to the data and client required.
@@ -232,6 +232,10 @@ func (o *createChangefeedOptions) completeCfg(
 		}
 	}
 
+	if o.commonChangefeedOptions.schemaRegistry != "" {
+		cfg.Sink.SchemaRegistry = o.commonChangefeedOptions.schemaRegistry
+	}
+
 	switch o.commonChangefeedOptions.sortEngine {
 	case model.SortInMemory:
 	case model.SortInFile:
@@ -271,7 +275,6 @@ func (o *createChangefeedOptions) completeCfg(
 			// TODO(neil) enable ID bucket.
 		}
 	}
-	cfg.SchemaRegistry = o.schemaRegistry
 	// Complete cfg.
 	o.cfg = cfg
 

pkg/cmd/cli/cli_changefeed_update.go

@@ -162,6 +162,8 @@ func (o *updateChangefeedOptions) applyChanges(oldInfo *model.ChangeFeedInfo, cm
 			if err = o.commonChangefeedOptions.strictDecodeConfig("TiCDC changefeed", cfg); err != nil {
 				log.Error("decode config file error", zap.Error(err))
 			}
+		case "schema-registry":
+			newInfo.Config.Sink.SchemaRegistry = o.commonChangefeedOptions.schemaRegistry
 		case "opts":
 			for _, opt := range o.commonChangefeedOptions.opts {
 				s := strings.SplitN(opt, "=", 2)

pkg/cmd/cli/cli_changefeed_update_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/pingcap/check"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/util/testleak"
 )
 
@@ -68,7 +69,25 @@ func (s *changefeedUpdateSuite) TestApplyChanges(c *check.C) {
 	c.Assert(newInfo.SortDir, check.Equals, ".")
 	file, err := os.ReadFile(filename)
 	c.Assert(err, check.IsNil)
-	c.Assert(strings.Contains(string(file), "this flag cannot be updated and will be ignored"), check.IsTrue)
+	c.Assert(
+		strings.Contains(string(file), "this flag cannot be updated and will be ignored"),
+		check.IsTrue,
+	)
+
+	// Test schema registry update
+	oldInfo = &model.ChangeFeedInfo{Config: config.GetDefaultReplicaConfig()}
+	c.Assert(oldInfo.Config.Sink.SchemaRegistry, check.Equals, "")
+	c.Assert(
+		cmd.ParseFlags([]string{"--schema-registry=https://username:password@localhost:8081"}),
+		check.IsNil,
+	)
+	newInfo, err = o.applyChanges(oldInfo, cmd)
+	c.Assert(err, check.IsNil)
+	c.Assert(
+		newInfo.Config.Sink.SchemaRegistry,
+		check.Equals,
+		"https://username:password@localhost:8081",
+	)
 }
 
 func initTestLogger(filename string) (func(), error) {

pkg/config/config_test_data.go

@@ -167,7 +167,8 @@ const (
           "b"
         ]
       }
-    ]
+    ],
+    "schema-registry": ""
   },
   "cyclic-replication": {
     "enable": false,
@@ -181,8 +182,7 @@ const (
     "max-log-size": 64,
     "flush-interval": 1000,
     "storage": ""
-  },
-  "schema-registry": ""
+  }
 }`
 
 	testCfgTestReplicaConfigMarshal2 = `{

pkg/config/replica_config.go

@@ -46,7 +46,6 @@ var defaultReplicaConfig = &ReplicaConfig{
 		FlushIntervalInMs: 1000,
 		Storage:           "",
 	},
-	SchemaRegistry: "",
 }
 
 // ReplicaConfig represents some addition replication config for a changefeed
@@ -62,7 +61,6 @@ type replicaConfig struct {
 	Sink             *SinkConfig       `toml:"sink" json:"sink"`
 	Cyclic           *CyclicConfig     `toml:"cyclic-replication" json:"cyclic-replication"`
 	Consistent       *ConsistentConfig `toml:"consistent" json:"consistent"`
-	SchemaRegistry   string            `toml:"schema-registry" json:"schema-registry"`
 }
 
 // Marshal returns the json marshal format of a ReplicationConfig

pkg/config/sink.go

@@ -37,6 +37,7 @@ type SinkConfig struct {
 	DispatchRules   []*DispatchRule   `toml:"dispatchers" json:"dispatchers"`
 	Protocol        string            `toml:"protocol" json:"protocol"`
 	ColumnSelectors []*ColumnSelector `toml:"column-selectors" json:"column-selectors"`
+	SchemaRegistry  string            `toml:"schema-registry" json:"schema-registry"`
 }
 
 // DispatchRule represents partition rule for a table