config(ticdc): Add alias for "dispatcher" in dispatch rules

pkg/config/replica_config.go

@@ -117,8 +117,8 @@ func (c *replicaConfig) fillFromV1(v1 *outdated.ReplicaConfigV1) {
 	}
 	for _, dispatch := range v1.Sink.DispatchRules {
 		c.Sink.DispatchRules = append(c.Sink.DispatchRules, &DispatchRule{
-			Matcher:       []string{fmt.Sprintf("%s.%s", dispatch.Schema, dispatch.Name)},
-			PartitionRule: dispatch.Rule,
+			Matcher:        []string{fmt.Sprintf("%s.%s", dispatch.Schema, dispatch.Name)},
+			DispatcherRule: dispatch.Rule,
 		})
 	}
 }

pkg/config/replica_config_test.go

@@ -77,9 +77,9 @@ func TestReplicaConfigOutDated(t *testing.T) {
 	conf.Mounter.WorkerNum = 3
 	conf.Sink.Protocol = "open-protocol"
 	conf.Sink.DispatchRules = []*DispatchRule{
-		{Matcher: []string{"a.b"}, PartitionRule: "r1"},
-		{Matcher: []string{"a.c"}, PartitionRule: "r2"},
-		{Matcher: []string{"a.d"}, PartitionRule: "r2"},
+		{Matcher: []string{"a.b"}, DispatcherRule: "r1"},
+		{Matcher: []string{"a.c"}, DispatcherRule: "r2"},
+		{Matcher: []string{"a.d"}, DispatcherRule: "r2"},
 	}
 	require.Equal(t, conf, conf2)
 }
@@ -94,4 +94,24 @@ func TestReplicaConfigValidate(t *testing.T) {
 	conf.Sink.Protocol = "canal"
 	conf.EnableOldValue = false
 	require.Regexp(t, ".*canal protocol requires old value to be enabled.*", conf.Validate())
+
+	conf = GetDefaultReplicaConfig()
+	conf.Sink.DispatchRules = []*DispatchRule{
+		{Matcher: []string{"a.b"}, DispatcherRule: "d1", PartitionRule: "r1"},
+	}
+	require.Regexp(t, ".*dispatcher and partition cannot be configured both.*", conf.Validate())
+
+	// Correct sink configuration.
+	conf = GetDefaultReplicaConfig()
+	conf.Sink.DispatchRules = []*DispatchRule{
+		{Matcher: []string{"a.b"}, DispatcherRule: "d1"},
+		{Matcher: []string{"a.c"}, PartitionRule: "p1"},
+		{Matcher: []string{"a.d"}},
+	}
+	err := conf.Validate()
+	require.Nil(t, err)
+	rules := conf.Sink.DispatchRules
+	require.Equal(t, "d1", rules[0].PartitionRule)
+	require.Equal(t, "p1", rules[1].PartitionRule)
+	require.Equal(t, "", rules[2].PartitionRule)
 }

pkg/config/sink.go

@@ -19,6 +19,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
+	"go.uber.org/zap"
 )
 
 // DefaultMaxMessageBytes sets the default value for max-message-bytes
@@ -40,9 +41,10 @@ type SinkConfig struct {
 
 // DispatchRule represents partition rule for a table
 type DispatchRule struct {
-	Matcher       []string `toml:"matcher" json:"matcher"`
-	PartitionRule string   `toml:"dispatcher" json:"dispatcher"`
-	TopicRule     string   `toml:"topic" json:"topic"`
+	Matcher        []string `toml:"matcher" json:"matcher"`
+	DispatcherRule string   `toml:"dispatcher" json:"dispatcher"`
+	PartitionRule  string   `toml:"partition" json:"partition"`
+	TopicRule      string   `toml:"topic" json:"topic"`
 }
 
 // ColumnSelector represents a column selector for a table.
@@ -62,6 +64,16 @@ func (s *SinkConfig) validate(enableOldValue bool) error {
 			}
 		}
 	}
+	for _, rule := range s.DispatchRules {
+		if rule.DispatcherRule != "" && rule.PartitionRule != "" {
+			log.Error("dispatcher and partition cannot be configured both", zap.Any("rule", rule))
+			return cerror.WrapError(cerror.ErrSinkInvalidConfig,
+				errors.New(fmt.Sprintf("dispatcher and partition cannot be configured both for rule:%v", rule)))
+		}
+		if rule.DispatcherRule != "" {
+			rule.PartitionRule = rule.DispatcherRule
+		}
+	}
 
 	return nil
 }