sink: add debezium-avro protocol

.golangci.yml

@@ -84,9 +84,11 @@ linters:
         - name: var-naming
 
     # G104: Audit errors not checked (duplicates errcheck).
+    # G115: integer overflow conversions are too noisy for this codebase.
     gosec:
       excludes:
         - G104
+        - G115
 
     # ST1000: don't require package comments.
     # ST1003: don't enforce naming conventions on legacy identifiers.

api/v2/changefeed.go

@@ -1770,7 +1770,8 @@ func verifyTable4MQ(
 		return nil
 	}
 
-	eventRouter, err := eventrouter.NewEventRouter(replicaConfig.Sink, topic, config.IsPulsarScheme(scheme), protocol == config.ProtocolAvro)
+	isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro
+	eventRouter, err := eventrouter.NewEventRouter(replicaConfig.Sink, topic, config.IsPulsarScheme(scheme), isAvroLike)
 	if err != nil {
 		return err
 	}

cmd/kafka-consumer/consumer.go

@@ -42,28 +42,48 @@ func getPartitionNum(o *option) (int32, error) {
 	}
 	defer admin.Close()
 
+	topics := strings.Split(o.topic, ",")
+	maxPartitionNum := int32(0)
 	timeout := 3000
-	for i := 0; i <= 30; i++ {
-		resp, err := admin.GetMetadata(&o.topic, false, timeout)
-		if err != nil {
-			if err.(kafka.Error).Code() == kafka.ErrTransport {
-				log.Info("retry get partition number", zap.Int("retryTime", i), zap.Int("timeout", timeout))
-				timeout += 100
-				continue
+	for _, topic := range topics {
+		topic = strings.TrimSpace(topic)
+		if topic == "" {
+			continue
+		}
+		found := false
+		for i := 0; i <= 30; i++ {
+			resp, err := admin.GetMetadata(&topic, false, timeout)
+			if err != nil {
+				var kafkaErr kafka.Error
+				if errors.As(err, &kafkaErr) && kafkaErr.Code() == kafka.ErrTransport {
+					log.Info("retry get partition number", zap.String("topic", topic), zap.Int("retryTime", i), zap.Int("timeout", timeout))
+					timeout += 100
+					continue
+				}
+				return 0, errors.Trace(err)
+			}
+			if topicDetail, ok := resp.Topics[topic]; ok {
+				numPartitions := int32(len(topicDetail.Partitions))
+				log.Info("get partition number of topic",
+					zap.String("topic", topic),
+					zap.Int32("partitionNum", numPartitions))
+				if numPartitions > maxPartitionNum {
+					maxPartitionNum = numPartitions
+				}
+				found = true
+				break
 			}
-			return 0, errors.Trace(err)
+			log.Info("retry get partition number", zap.String("topic", topic))
+			time.Sleep(1 * time.Second)
 		}
-		if topicDetail, ok := resp.Topics[o.topic]; ok {
-			numPartitions := int32(len(topicDetail.Partitions))
-			log.Info("get partition number of topic",
-				zap.String("topic", o.topic),
-				zap.Int32("partitionNum", numPartitions))
-			return numPartitions, nil
+		if !found {
+			return 0, errors.Errorf("get partition number(%s) timeout", topic)
 		}
-		log.Info("retry get partition number", zap.String("topic", o.topic))
-		time.Sleep(1 * time.Second)
 	}
-	return 0, errors.Errorf("get partition number(%s) timeout", o.topic)
+	if maxPartitionNum == 0 {
+		return 0, errors.Errorf("get partition number(%s) timeout", o.topic)
+	}
+	return maxPartitionNum, nil
 }
 
 type consumer struct {

cmd/kafka-consumer/option.go

@@ -122,11 +122,11 @@ func (o *option) Adjust(upstreamURIStr string, configFile string) {
 		}
 		o.partitionNum = int32(c)
 	}
-	partitionNum, err := getPartitionNum(o)
-	if err != nil {
-		log.Panic("cannot get the partition number", zap.String("topic", o.topic), zap.Error(err))
-	}
 	if o.partitionNum == 0 {
+		partitionNum, err := getPartitionNum(o)
+		if err != nil {
+			log.Panic("cannot get the partition number", zap.String("topic", o.topic), zap.Error(err))
+		}
 		o.partitionNum = partitionNum
 	}
 

cmd/kafka-consumer/writer.go

@@ -119,7 +119,8 @@ func newWriter(ctx context.Context, o *option) *writer {
 		w.progresses[i] = newPartitionProgress(int32(i), decoder)
 	}
 
-	eventRouter, err := eventrouter.NewEventRouter(o.sinkConfig, o.topic, false, o.protocol == config.ProtocolAvro)
+	isAvroLike := o.protocol == config.ProtocolAvro || o.protocol == config.ProtocolDebeziumAvro
+	eventRouter, err := eventrouter.NewEventRouter(o.sinkConfig, o.topic, false, isAvroLike)
 	if err != nil {
 		log.Panic("initialize the event router failed",
 			zap.Any("protocol", o.protocol), zap.Any("topic", o.topic),
@@ -493,7 +494,8 @@ func (w *writer) onDDL(ddl *event.DDLEvent) {
 		return
 	}
 	switch w.protocol {
-	case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro, config.ProtocolSimple, config.ProtocolDebezium:
+	case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro, config.ProtocolSimple,
+		config.ProtocolDebezium, config.ProtocolDebeziumAvro:
 	default:
 		return
 	}
@@ -632,7 +634,8 @@ func (w *writer) appendRow2Group(dml *event.DMLEvent, progress *partitionProgres
 			zap.Stringer("eventType", dml.RowTypes[0]),
 			// zap.Any("columns", row.Columns), zap.Any("preColumns", row.PreColumns),
 			zap.Any("protocol", w.protocol), zap.Bool("IsPartition", dml.TableInfo.TableName.IsPartition))
-	case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro, config.ProtocolDebezium:
+	case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro,
+		config.ProtocolDebezium, config.ProtocolDebeziumAvro:
 		// for partition table, these protocols cannot assign physical table id to each dml message,
 		// we cannot distinguish whether it's a real fallback event or not, still append it.
 		if w.partitionTableAccessor.IsPartitionTable(schema, table) {

cmd/kafka-consumer/writer_test.go

@@ -320,44 +320,51 @@ func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) {
 }
 
 func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) {
-	replicaCfg := config.GetDefaultReplicaConfig()
-	eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false)
-	require.NoError(t, err)
+	for _, protocol := range []config.Protocol{
+		config.ProtocolDebezium,
+		config.ProtocolDebeziumAvro,
+	} {
+		t.Run(protocol.String(), func(t *testing.T) {
+			replicaCfg := config.GetDefaultReplicaConfig()
+			eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false)
+			require.NoError(t, err)
 
-	w := &writer{
-		progresses:             []*partitionProgress{{partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}},
-		eventRouter:            eventRouter,
-		protocol:               config.ProtocolDebezium,
-		partitionTableAccessor: codeccommon.NewPartitionTableAccessor(),
-	}
+			w := &writer{
+				progresses:             []*partitionProgress{{partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}},
+				eventRouter:            eventRouter,
+				protocol:               protocol,
+				partitionTableAccessor: codeccommon.NewPartitionTableAccessor(),
+			}
 
-	w.partitionTableAccessor.Add("target", "src")
-	ddl := &commonEvent.DDLEvent{
-		Query:      "CREATE TABLE `target`.`dst` LIKE `target`.`src`",
-		SchemaName: "target",
-		TableName:  "dst",
-		Type:       byte(timodel.ActionCreateTable),
-	}
-	w.onDDL(ddl)
-	require.True(t, w.partitionTableAccessor.IsPartitionTable("target", "dst"))
+			w.partitionTableAccessor.Add("target", "src")
+			ddl := &commonEvent.DDLEvent{
+				Query:      "CREATE TABLE `target`.`dst` LIKE `target`.`src`",
+				SchemaName: "target",
+				TableName:  "dst",
+				Type:       byte(timodel.ActionCreateTable),
+			}
+			w.onDDL(ddl)
+			require.True(t, w.partitionTableAccessor.IsPartitionTable("target", "dst"))
 
-	newDMLEvent := func(commitTs uint64) *commonEvent.DMLEvent {
-		return &commonEvent.DMLEvent{
-			PhysicalTableID: 1,
-			CommitTs:        commitTs,
-			RowTypes:        []common.RowType{common.RowTypeUpdate},
-			Rows:            chunk.NewChunkWithCapacity(nil, 0),
-			TableInfo: &common.TableInfo{
-				TableName: common.TableName{Schema: "target", Table: "dst"},
-			},
-		}
-	}
+			newDMLEvent := func(commitTs uint64) *commonEvent.DMLEvent {
+				return &commonEvent.DMLEvent{
+					PhysicalTableID: 1,
+					CommitTs:        commitTs,
+					RowTypes:        []common.RowType{common.RowTypeUpdate},
+					Rows:            chunk.NewChunkWithCapacity(nil, 0),
+					TableInfo: &common.TableInfo{
+						TableName: common.TableName{Schema: "target", Table: "dst"},
+					},
+				}
+			}
 
-	progress := w.progresses[0]
-	w.appendRow2Group(newDMLEvent(200), progress, kafka.Offset(10))
-	w.appendRow2Group(newDMLEvent(100), progress, kafka.Offset(11))
+			progress := w.progresses[0]
+			w.appendRow2Group(newDMLEvent(200), progress, kafka.Offset(10))
+			w.appendRow2Group(newDMLEvent(100), progress, kafka.Offset(11))
 
-	resolved := progress.eventsGroup[1].ResolveInto(150, nil)
-	require.Len(t, resolved, 1)
-	require.Equal(t, uint64(100), resolved[0].CommitTs)
+			resolved := progress.eventsGroup[1].ResolveInto(150, nil)
+			require.Len(t, resolved, 1)
+			require.Equal(t, uint64(100), resolved[0].CommitTs)
+		})
+	}
 }

downstreamadapter/sink/helper/helper.go

@@ -100,7 +100,7 @@ func GetProtocol(protocolStr string) (config.Protocol, error) {
 // GetFileExtension returns the extension for specific protocol
 func GetFileExtension(protocol config.Protocol) string {
 	switch protocol {
-	case config.ProtocolAvro, config.ProtocolCanalJSON, config.ProtocolMaxwell,
+	case config.ProtocolAvro, config.ProtocolDebeziumAvro, config.ProtocolCanalJSON, config.ProtocolMaxwell,
 		config.ProtocolOpen, config.ProtocolSimple:
 		return ".json"
 	case config.ProtocolCraft:

downstreamadapter/sink/kafka/helper.go

@@ -77,8 +77,9 @@ func newKafkaSinkComponentWithFactory(ctx context.Context,
 		return kafkaComponent, protocol, errors.WrapError(errors.ErrKafkaNewProducer, err)
 	}
 
+	isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro
 	kafkaComponent.eventRouter, err = eventrouter.NewEventRouter(
-		sinkConfig, topic, false, protocol == config.ProtocolAvro)
+		sinkConfig, topic, false, isAvroLike)
 	if err != nil {
 		return kafkaComponent, protocol, errors.Trace(err)
 	}

pkg/config/changefeed.go

@@ -491,8 +491,9 @@ func (info *ChangeFeedInfo) RmUnusedFields() {
 		info.rmMQOnlyFields()
 	} else {
 		// remove schema registry for MQ downstream with
-		// protocol other than avro
-		if util.GetOrZero(info.Config.Sink.Protocol) != ProtocolAvro.String() {
+		// protocol other than avro or debezium-avro
+		protocol := util.GetOrZero(info.Config.Sink.Protocol)
+		if protocol != ProtocolAvro.String() && protocol != ProtocolDebeziumAvro.String() {
 			info.Config.Sink.SchemaRegistry = nil
 		}
 	}

pkg/config/changefeed_test.go

@@ -65,3 +65,38 @@ func TestChangeFeedInfoToChangefeedConfigBatchFields(t *testing.T) {
 	assertBatchFields(util.AddressOf(0), util.AddressOf(0))
 	assertBatchFields(util.AddressOf(123), util.AddressOf(456))
 }
+
+func TestChangeFeedInfoRmUnusedFieldsKeepsSchemaRegistryForAvroProtocols(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		protocol     Protocol
+		keepRegistry bool
+	}{
+		{protocol: ProtocolAvro, keepRegistry: true},
+		{protocol: ProtocolDebeziumAvro, keepRegistry: true},
+		{protocol: ProtocolDebezium, keepRegistry: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.protocol.String(), func(t *testing.T) {
+			t.Parallel()
+
+			cfg := GetDefaultReplicaConfig()
+			cfg.Sink.Protocol = util.AddressOf(tt.protocol.String())
+			cfg.Sink.SchemaRegistry = util.AddressOf("http://127.0.0.1:8088")
+			info := &ChangeFeedInfo{
+				SinkURI: "kafka://127.0.0.1:9092/topic",
+				Config:  cfg,
+			}
+
+			info.RmUnusedFields()
+			if tt.keepRegistry {
+				require.NotNil(t, info.Config.Sink.SchemaRegistry)
+				require.Equal(t, "http://127.0.0.1:8088", *info.Config.Sink.SchemaRegistry)
+			} else {
+				require.Nil(t, info.Config.Sink.SchemaRegistry)
+			}
+		})
+	}
+}

pkg/config/sink.go

@@ -144,7 +144,8 @@ type SinkConfig struct {
 	DispatchRules []*DispatchRule `toml:"dispatchers" json:"dispatchers,omitempty"`
 
 	ColumnSelectors []*ColumnSelector `toml:"column-selectors" json:"column-selectors,omitempty"`
-	// SchemaRegistry is only available when the downstream is MQ using avro protocol.
+	// SchemaRegistry is only available when the downstream is MQ using avro protocol
+	// or debezium protocol with Confluent Avro encoding.
 	SchemaRegistry *string `toml:"schema-registry" json:"schema-registry,omitempty"`
 	// EncoderConcurrency is only available when the downstream is MQ.
 	EncoderConcurrency *int `toml:"encoder-concurrency" json:"encoder-concurrency,omitempty"`
@@ -965,7 +966,7 @@ func (s *SinkConfig) ValidateProtocol(scheme string) error {
 		if s.OpenProtocol != nil {
 			outputOldValue = s.OpenProtocol.OutputOldValue
 		}
-	case ProtocolDebezium:
+	case ProtocolDebezium, ProtocolDebeziumAvro:
 		if s.Debezium != nil {
 			outputOldValue = s.Debezium.OutputOldValue
 		}

pkg/config/sink_protocol.go

@@ -41,6 +41,7 @@ const (
 	ProtocolCsv
 	ProtocolDebezium
 	ProtocolSimple
+	ProtocolDebeziumAvro
 )
 
 // IsBatchEncode returns whether the protocol is a batch encoder.
@@ -71,6 +72,8 @@ func ParseSinkProtocolFromString(protocol string) (Protocol, error) {
 		return ProtocolCsv, nil
 	case "debezium":
 		return ProtocolDebezium, nil
+	case "debezium-avro":
+		return ProtocolDebeziumAvro, nil
 	case "simple":
 		return ProtocolSimple, nil
 	default:
@@ -101,6 +104,8 @@ func (p Protocol) String() string {
 		return "debezium"
 	case ProtocolSimple:
 		return "simple"
+	case ProtocolDebeziumAvro:
+		return "debezium-avro"
 	default:
 		panic("unreachable")
 	}

pkg/config/sink_protocol_test.go

@@ -62,6 +62,10 @@ func TestParseSinkProtocolFromString(t *testing.T) {
 			protocol:             "open-protocol",
 			expectedProtocolEnum: ProtocolOpen,
 		},
+		{
+			protocol:             "debezium-avro",
+			expectedProtocolEnum: ProtocolDebeziumAvro,
+		},
 	}
 
 	for _, tc := range testCases {
@@ -109,6 +113,10 @@ func TestString(t *testing.T) {
 			protocolEnum:     ProtocolOpen,
 			expectedProtocol: "open-protocol",
 		},
+		{
+			protocolEnum:     ProtocolDebeziumAvro,
+			expectedProtocol: "debezium-avro",
+		},
 	}
 
 	for _, tc := range testCases {
@@ -151,6 +159,10 @@ func TestIsBatchEncoder(t *testing.T) {
 			protocolEnum: ProtocolOpen,
 			expect:       true,
 		},
+		{
+			protocolEnum: ProtocolDebeziumAvro,
+			expect:       false,
+		},
 	}
 
 	for _, tc := range testCases {

pkg/sink/codec/builder.go

@@ -41,6 +41,8 @@ func NewEventEncoder(ctx context.Context, cfg *common.Config) (common.EventEncod
 		return canal.NewJSONRowEventEncoder(ctx, cfg)
 	case config.ProtocolDebezium:
 		return debezium.NewBatchEncoder(cfg, config.GetGlobalServerConfig().ClusterID), nil
+	case config.ProtocolDebeziumAvro:
+		return debezium.NewAvroBatchEncoder(ctx, cfg, config.GetGlobalServerConfig().ClusterID)
 	case config.ProtocolSimple:
 		return simple.NewEncoder(ctx, cfg)
 	default:
@@ -67,6 +69,8 @@ func NewEventDecoder(
 		return simple.NewDecoder(ctx, codecConfig, upstreamTiDB)
 	case config.ProtocolDebezium:
 		return debezium.NewDecoder(codecConfig, idx, upstreamTiDB), nil
+	case config.ProtocolDebeziumAvro:
+		return debezium.NewAvroDecoder(ctx, codecConfig, idx, upstreamTiDB)
 	default:
 	}
 	log.Panic("Protocol not supported", zap.Any("Protocol", codecConfig.Protocol))