cdc: Add Debezium protocol

cdc/sink/util/helper.go

@@ -24,6 +24,7 @@ import (
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 	"github.com/pingcap/tiflow/pkg/sink/kafka"
+	"github.com/pingcap/tiflow/pkg/util"
 )
 
 // GetTopic returns the topic name from the sink URI.
@@ -79,7 +80,15 @@ func GetEncoderConfig(
 	// Always set encoder's `MaxMessageBytes` equal to producer's `MaxMessageBytes`
 	// to prevent that the encoder generate batched message too large
 	// then cause producer meet `message too large`.
-	encoderConfig = encoderConfig.WithMaxMessageBytes(maxMsgBytes).WithChangefeedID(changefeedID)
+	encoderConfig = encoderConfig.
+		WithMaxMessageBytes(maxMsgBytes).
+		WithChangefeedID(changefeedID)
+
+	tz, err := util.GetTimezone(config.GetGlobalServerConfig().TZ)
+	if err != nil {
+		return nil, cerror.WrapError(cerror.ErrSinkInvalidConfig, err)
+	}
+	encoderConfig.TimeZone = tz
 
 	if err := encoderConfig.Validate(); err != nil {
 		return nil, cerror.WrapError(cerror.ErrSinkInvalidConfig, err)

pkg/config/sink_protocol.go

@@ -38,6 +38,7 @@ const (
 	ProtocolCraft
 	ProtocolOpen
 	ProtocolCsv
+	ProtocolDebezium
 	ProtocolSimple
 )
 
@@ -67,6 +68,8 @@ func ParseSinkProtocolFromString(protocol string) (Protocol, error) {
 		return ProtocolOpen, nil
 	case "csv":
 		return ProtocolCsv, nil
+	case "debezium":
+		return ProtocolDebezium, nil
 	case "simple":
 		return ProtocolSimple, nil
 	default:
@@ -93,6 +96,8 @@ func (p Protocol) String() string {
 		return "open-protocol"
 	case ProtocolCsv:
 		return "csv"
+	case ProtocolDebezium:
+		return "debezium"
 	case ProtocolSimple:
 		return "simple"
 	default:

pkg/errors/cdc_errors.go

@@ -416,6 +416,10 @@ var (
 		"csv decode failed",
 		errors.RFCCodeText("CDC:ErrCSVDecodeFailed"),
 	)
+	ErrDebeziumEncodeFailed = errors.Normalize(
+		"debezium encode failed",
+		errors.RFCCodeText("CDC:ErrDebeziumEncodeFailed"),
+	)
 	ErrStorageSinkInvalidConfig = errors.Normalize(
 		"storage sink config invalid",
 		errors.RFCCodeText("CDC:ErrStorageSinkInvalidConfig"),

pkg/sink/codec/builder/encoder_builder.go

@@ -24,14 +24,16 @@ import (
 	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 	"github.com/pingcap/tiflow/pkg/sink/codec/craft"
 	"github.com/pingcap/tiflow/pkg/sink/codec/csv"
+	"github.com/pingcap/tiflow/pkg/sink/codec/debezium"
 	"github.com/pingcap/tiflow/pkg/sink/codec/maxwell"
 	"github.com/pingcap/tiflow/pkg/sink/codec/open"
 	"github.com/pingcap/tiflow/pkg/sink/codec/simple"
 )
 
 // NewRowEventEncoderBuilder returns an RowEventEncoderBuilder
 func NewRowEventEncoderBuilder(
-	ctx context.Context, cfg *common.Config,
+	ctx context.Context,
+	cfg *common.Config,
 ) (codec.RowEventEncoderBuilder, error) {
 	switch cfg.Protocol {
 	case config.ProtocolDefault, config.ProtocolOpen:
@@ -46,6 +48,8 @@ func NewRowEventEncoderBuilder(
 		return canal.NewJSONRowEventEncoderBuilder(ctx, cfg)
 	case config.ProtocolCraft:
 		return craft.NewBatchEncoderBuilder(cfg), nil
+	case config.ProtocolDebezium:
+		return debezium.NewBatchEncoderBuilder(cfg), nil
 	case config.ProtocolSimple:
 		return simple.NewBuilder(cfg), nil
 	default:

pkg/sink/codec/common/config.go

@@ -16,6 +16,7 @@ package common
 import (
 	"net/http"
 	"net/url"
+	"time"
 
 	"github.com/gin-gonic/gin/binding"
 	"github.com/imdario/mergo"
@@ -73,6 +74,9 @@ type Config struct {
 
 	// for open protocol
 	OnlyOutputUpdatedColumns bool
+
+	// Currently only Debezium protocol is aware of the time zone
+	TimeZone *time.Location
 }
 
 // NewConfig return a Config for codec
@@ -94,6 +98,8 @@ func NewConfig(protocol config.Protocol) *Config {
 		OnlyOutputUpdatedColumns:   false,
 		DeleteOnlyHandleKeyColumns: false,
 		LargeMessageHandle:         config.NewDefaultLargeMessageHandleConfig(),
+
+		TimeZone: time.Local,
 	}
 }