config (ticdc): Fix sink protocol can no specify by config file. (#6361…

…) (#7508)

close #6360, ref #7501

pkg/config/sink.go

@@ -155,7 +155,18 @@ func (s *SinkConfig) applyParameter(sinkURI *url.URL) error {
 		return cerror.ErrSinkURIInvalid.GenWithStackByArgs(errMsg)
 	}
 
-	s.Protocol = params.Get(ProtocolKey)
+	protocolFromURI := params.Get(ProtocolKey)
+	if protocolFromURI != "" {
+		if s.Protocol != "" {
+			log.Warn(
+				fmt.Sprintf("protocol is specified in both sink URI and config file"+
+					"the value in sink URI will be used"+
+					"protocol in sink URI:%s, protocol in config file:%s",
+					protocolFromURI, s.Protocol))
+		}
+		s.Protocol = protocolFromURI
+	}
+
 	// validate that protocol is compatible with the scheme
 	if IsMqScheme(sinkURI.Scheme) {
 		var protocol Protocol

pkg/config/sink_test.go

@@ -159,3 +159,38 @@ func TestValidateApplyParameter(t *testing.T) {
 		}
 	}
 }
+
+func TestApplyParameter(t *testing.T) {
+	t.Parallel()
+	testCases := []struct {
+		sinkConfig *SinkConfig
+		sinkURI    string
+		result     string
+	}{
+		{
+			sinkConfig: &SinkConfig{
+				Protocol: "default",
+			},
+			sinkURI: "kafka://127.0.0.1:9092?protocol=whatever",
+			result:  "whatever",
+		},
+		{
+			sinkConfig: &SinkConfig{},
+			sinkURI:    "kafka://127.0.0.1:9092?protocol=default",
+			result:     "default",
+		},
+		{
+			sinkConfig: &SinkConfig{
+				Protocol: "default",
+			},
+			sinkURI: "kafka://127.0.0.1:9092",
+			result:  "default",
+		},
+	}
+	for _, c := range testCases {
+		parsedSinkURI, err := url.Parse(c.sinkURI)
+		require.Nil(t, err)
+		c.sinkConfig.applyParameter(parsedSinkURI)
+		require.Equal(t, c.result, c.sinkConfig.Protocol)
+	}
+}