codec(ticdc): open protocol decoder support handle only message (ping…

…cap#9454) close pingcap#9443
3AceShowHand · Aug 29, 2023 · a2aee0c · a2aee0c
1 parent 1a883bd
commit a2aee0c
Show file tree

Hide file tree

Showing 11 changed files with 405 additions and 227 deletions.
diff --git a/cmd/kafka-consumer/main.go b/cmd/kafka-consumer/main.go
@@ -586,7 +586,7 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 
 	switch c.codecConfig.Protocol {
 	case config.ProtocolOpen, config.ProtocolDefault:
-		decoder, err = open.NewBatchDecoder(ctx, c.codecConfig)
+		decoder, err = open.NewBatchDecoder(ctx, c.codecConfig, c.upstreamTiDB)
 	case config.ProtocolCanalJSON:
 		decoder, err = canal.NewBatchDecoder(ctx, c.codecConfig, c.upstreamTiDB)
 		if err != nil {

diff --git a/pkg/sink/codec/builder/codec_test.go b/pkg/sink/codec/builder/codec_test.go
@@ -306,7 +306,7 @@ func BenchmarkJsonDecoding(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		for _, message := range codecJSONEncodedRowChanges {
 			codecConfig := common.NewConfig(config.ProtocolOpen)
-			decoder, err := open.NewBatchDecoder(context.Background(), codecConfig)
+			decoder, err := open.NewBatchDecoder(context.Background(), codecConfig, nil)
 			require.NoError(b, err)
 			if err := decoder.AddKeyValue(message.Key, message.Value); err != nil {
 				panic(err)