This is an automated cherry-pick of #3747

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>
pingcap · Dec 8, 2021 · 279cb04 · 279cb04
1 parent 81948a8
commit 279cb04
Show file tree

Hide file tree

Showing 16 changed files with 627 additions and 81 deletions.
diff --git a/cdc/sink/codec/avro.go b/cdc/sink/codec/avro.go
@@ -27,6 +27,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/cdc/model"
+	"github.com/pingcap/ticdc/pkg/config"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/security"
 	"github.com/pingcap/ticdc/pkg/util"
@@ -88,7 +89,7 @@ func (a *AvroEventBatchEncoder) SetTimeZone(tz *time.Location) {
 // AppendRowChangedEvent appends a row change event to the encoder
 // NOTE: the encoder can only store one RowChangedEvent!
 func (a *AvroEventBatchEncoder) AppendRowChangedEvent(e *model.RowChangedEvent) (EncoderResult, error) {
-	mqMessage := NewMQMessage(ProtocolAvro, nil, nil, e.CommitTs, model.MqMessageTypeRow, &e.Table.Schema, &e.Table.Table)
+	mqMessage := NewMQMessage(config.ProtocolAvro, nil, nil, e.CommitTs, model.MqMessageTypeRow, &e.Table.Schema, &e.Table.Table)
 
 	if !e.IsDelete() {
 		res, err := avroEncode(e.Table, a.valueSchemaManager, e.TableInfoVersion, e.Columns, a.tz)

diff --git a/cdc/sink/codec/canal.go b/cdc/sink/codec/canal.go
@@ -23,6 +23,7 @@ import (
 	"github.com/golang/protobuf/proto" // nolint:staticcheck
 	"github.com/pingcap/errors"
 	"github.com/pingcap/ticdc/cdc/model"
+	"github.com/pingcap/ticdc/pkg/config"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
 	canal "github.com/pingcap/ticdc/proto/canal"
 	mm "github.com/pingcap/tidb/parser/model"
@@ -373,7 +374,7 @@ func (d *CanalEventBatchEncoder) EncodeDDLEvent(e *model.DDLEvent) (*MQMessage,
 		return nil, cerror.WrapError(cerror.ErrCanalEncodeFailed, err)
 	}
 
-	return newDDLMQMessage(ProtocolCanal, nil, b, e), nil
+	return newDDLMQMessage(config.ProtocolCanal, nil, b, e), nil
 }
 
 // Build implements the EventBatchEncoder interface
@@ -391,7 +392,7 @@ func (d *CanalEventBatchEncoder) Build() []*MQMessage {
 	if err != nil {
 		log.Panic("Error when serializing Canal packet", zap.Error(err))
 	}
-	ret := NewMQMessage(ProtocolCanal, nil, value, 0, model.MqMessageTypeRow, nil, nil)
+	ret := NewMQMessage(config.ProtocolCanal, nil, value, 0, model.MqMessageTypeRow, nil, nil)
 	d.messages.Reset()
 	d.resetPacket()
 	return []*MQMessage{ret}

diff --git a/cdc/sink/codec/canal_flat.go b/cdc/sink/codec/canal_flat.go
@@ -20,6 +20,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/cdc/model"
+	"github.com/pingcap/ticdc/pkg/config"
 	cerrors "github.com/pingcap/ticdc/pkg/errors"
 	canal "github.com/pingcap/ticdc/proto/canal"
 	"go.uber.org/zap"
@@ -184,7 +185,20 @@ func (c *CanalFlatEventBatchEncoder) newFlatMessageForDDL(e *model.DDLEvent) *ca
 
 // EncodeCheckpointEvent is no-op
 func (c *CanalFlatEventBatchEncoder) EncodeCheckpointEvent(ts uint64) (*MQMessage, error) {
+<<<<<<< HEAD
 	return nil, nil
+=======
+	if !c.enableTiDBExtension {
+		return nil, nil
+	}
+
+	msg := c.newFlatMessage4CheckpointEvent(ts)
+	value, err := json.Marshal(msg)
+	if err != nil {
+		return nil, cerrors.WrapError(cerrors.ErrCanalEncodeFailed, err)
+	}
+	return newResolvedMQMessage(config.ProtocolCanalJSON, nil, value, ts), nil
+>>>>>>> bedf904e7 (config(ticdc): Fix old value configuration check for maxwell protocol (#3747))
 }
 
 // AppendRowChangedEvent implements the interface EventBatchEncoder
@@ -222,7 +236,7 @@ func (c *CanalFlatEventBatchEncoder) EncodeDDLEvent(e *model.DDLEvent) (*MQMessa
 	if err != nil {
 		return nil, cerrors.WrapError(cerrors.ErrCanalEncodeFailed, err)
 	}
-	return newDDLMQMessage(ProtocolCanalJSON, nil, value, e), nil
+	return newDDLMQMessage(config.ProtocolCanalJSON, nil, value, e), nil
 }
 
 // Build implements the EventBatchEncoder interface
@@ -237,7 +251,11 @@ func (c *CanalFlatEventBatchEncoder) Build() []*MQMessage {
 			log.Panic("CanalFlatEventBatchEncoder", zap.Error(err))
 			return nil
 		}
+<<<<<<< HEAD
 		ret[i] = NewMQMessage(ProtocolCanalJSON, nil, value, msg.tikvTs, model.MqMessageTypeRow, &msg.Schema, &msg.Table)
+=======
+		ret[i] = NewMQMessage(config.ProtocolCanalJSON, nil, value, msg.getTikvTs(), model.MqMessageTypeRow, msg.getSchema(), msg.getTable())
+>>>>>>> bedf904e7 (config(ticdc): Fix old value configuration check for maxwell protocol (#3747))
 	}
 	c.resolvedBuf = c.resolvedBuf[0:0]
 	return ret

diff --git a/cdc/sink/codec/craft.go b/cdc/sink/codec/craft.go
@@ -21,6 +21,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/ticdc/cdc/model"
 	"github.com/pingcap/ticdc/cdc/sink/codec/craft"
+	"github.com/pingcap/ticdc/pkg/config"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
 )
 
@@ -38,15 +39,15 @@ type CraftEventBatchEncoder struct {
 
 // EncodeCheckpointEvent implements the EventBatchEncoder interface
 func (e *CraftEventBatchEncoder) EncodeCheckpointEvent(ts uint64) (*MQMessage, error) {
-	return newResolvedMQMessage(ProtocolCraft, nil, craft.NewResolvedEventEncoder(e.allocator, ts).Encode(), ts), nil
+	return newResolvedMQMessage(config.ProtocolCraft, nil, craft.NewResolvedEventEncoder(e.allocator, ts).Encode(), ts), nil
 }
 
 func (e *CraftEventBatchEncoder) flush() {
 	headers := e.rowChangedBuffer.GetHeaders()
 	ts := headers.GetTs(0)
 	schema := headers.GetSchema(0)
 	table := headers.GetTable(0)
-	e.messageBuf = append(e.messageBuf, NewMQMessage(ProtocolCraft, nil, e.rowChangedBuffer.Encode(), ts, model.MqMessageTypeRow, &schema, &table))
+	e.messageBuf = append(e.messageBuf, NewMQMessage(config.ProtocolCraft, nil, e.rowChangedBuffer.Encode(), ts, model.MqMessageTypeRow, &schema, &table))
 }
 
 // AppendRowChangedEvent implements the EventBatchEncoder interface
@@ -65,7 +66,7 @@ func (e *CraftEventBatchEncoder) AppendResolvedEvent(ts uint64) (EncoderResult,
 
 // EncodeDDLEvent implements the EventBatchEncoder interface
 func (e *CraftEventBatchEncoder) EncodeDDLEvent(ev *model.DDLEvent) (*MQMessage, error) {
-	return newDDLMQMessage(ProtocolCraft, nil, craft.NewDDLEventEncoder(e.allocator, ev).Encode(), ev), nil
+	return newDDLMQMessage(config.ProtocolCraft, nil, craft.NewDDLEventEncoder(e.allocator, ev).Encode(), ev), nil
 }
 
 // Build implements the EventBatchEncoder interface

diff --git a/cdc/sink/codec/interface.go b/cdc/sink/codec/interface.go
@@ -16,11 +16,11 @@ package codec
 import (
 	"context"
 	"encoding/binary"
-	"strings"
 	"time"
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/cdc/model"
+	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/security"
 	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"
@@ -63,7 +63,7 @@ type MQMessage struct {
 	Schema   *string             // schema
 	Table    *string             // table
 	Type     model.MqMessageType // type
-	Protocol Protocol            // protocol
+	Protocol config.Protocol     // protocol
 }
 
 // maximumRecordOverhead is used to calculate ProducerMessage's byteSize by sarama kafka client.
@@ -83,17 +83,17 @@ func (m *MQMessage) PhysicalTime() time.Time {
 	return oracle.GetTimeFromTS(m.Ts)
 }
 
-func newDDLMQMessage(proto Protocol, key, value []byte, event *model.DDLEvent) *MQMessage {
+func newDDLMQMessage(proto config.Protocol, key, value []byte, event *model.DDLEvent) *MQMessage {
 	return NewMQMessage(proto, key, value, event.CommitTs, model.MqMessageTypeDDL, &event.TableInfo.Schema, &event.TableInfo.Table)
 }
 
-func newResolvedMQMessage(proto Protocol, key, value []byte, ts uint64) *MQMessage {
+func newResolvedMQMessage(proto config.Protocol, key, value []byte, ts uint64) *MQMessage {
 	return NewMQMessage(proto, key, value, ts, model.MqMessageTypeResolved, nil, nil)
 }
 
 // NewMQMessage should be used when creating a MQMessage struct.
 // It copies the input byte slices to avoid any surprises in asynchronous MQ writes.
-func NewMQMessage(proto Protocol, key []byte, value []byte, ts uint64, ty model.MqMessageType, schema, table *string) *MQMessage {
+func NewMQMessage(proto config.Protocol, key []byte, value []byte, ts uint64, ty model.MqMessageType, schema, table *string) *MQMessage {
 	ret := &MQMessage{
 		Key:      nil,
 		Value:    nil,
@@ -143,58 +143,24 @@ const (
 	EncoderNeedSyncWrite
 )
 
-// Protocol is the protocol of the mq message
-type Protocol int
-
-// Enum types of the Protocol
-const (
-	ProtocolDefault Protocol = iota
-	ProtocolCanal
-	ProtocolAvro
-	ProtocolMaxwell
-	ProtocolCanalJSON
-	ProtocolCraft
-)
-
-// FromString converts the protocol from string to Protocol enum type
-func (p *Protocol) FromString(protocol string) {
-	switch strings.ToLower(protocol) {
-	case "default":
-		*p = ProtocolDefault
-	case "canal":
-		*p = ProtocolCanal
-	case "avro":
-		*p = ProtocolAvro
-	case "maxwell":
-		*p = ProtocolMaxwell
-	case "canal-json":
-		*p = ProtocolCanalJSON
-	case "craft":
-		*p = ProtocolCraft
-	default:
-		*p = ProtocolDefault
-		log.Warn("can't support codec protocol, using default protocol", zap.String("protocol", protocol))
-	}
-}
-
 type EncoderBuilder interface {
 	Build(ctx context.Context) (EventBatchEncoder, error)
 }
 
 // NewEventBatchEncoderBuilder returns an EncoderBuilder
-func NewEventBatchEncoderBuilder(p Protocol, credential *security.Credential, opts map[string]string) (EncoderBuilder, error) {
+func NewEventBatchEncoderBuilder(p config.Protocol, credential *security.Credential, opts map[string]string) (EncoderBuilder, error) {
 	switch p {
-	case ProtocolDefault:
+	case config.ProtocolDefault:
 		return newJSONEventBatchEncoderBuilder(opts), nil
-	case ProtocolCanal:
+	case config.ProtocolCanal:
 		return newCanalEventBatchEncoderBuilder(opts), nil
-	case ProtocolAvro:
+	case config.ProtocolAvro:
 		return newAvroEventBatchEncoderBuilder(credential, opts)
-	case ProtocolMaxwell:
+	case config.ProtocolMaxwell:
 		return newMaxwellEventBatchEncoderBuilder(opts), nil
-	case ProtocolCanalJSON:
+	case config.ProtocolCanalJSON:
 		return newCanalFlatEventBatchEncoderBuilder(opts), nil
-	case ProtocolCraft:
+	case config.ProtocolCraft:
 		return newCraftEventBatchEncoderBuilder(opts), nil
 	default:
 		log.Warn("unknown codec protocol value of EventBatchEncoder, use open-protocol as the default", zap.Int("protocol_value", int(p)))

diff --git a/cdc/sink/codec/interface_test.go b/cdc/sink/codec/interface_test.go
@@ -16,6 +16,7 @@ package codec
 import (
 	"github.com/pingcap/check"
 	"github.com/pingcap/ticdc/cdc/model"
+	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/util/testleak"
 	timodel "github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
@@ -54,15 +55,15 @@ func (s *codecInterfaceSuite) TestCreate(c *check.C) {
 		CommitTs: 5678,
 	}
 
-	msg := NewMQMessage(ProtocolDefault, []byte("key1"), []byte("value1"), rowEvent.CommitTs, model.MqMessageTypeRow, &rowEvent.Table.Schema, &rowEvent.Table.Table)
+	msg := NewMQMessage(config.ProtocolDefault, []byte("key1"), []byte("value1"), rowEvent.CommitTs, model.MqMessageTypeRow, &rowEvent.Table.Schema, &rowEvent.Table.Table)
 
 	c.Assert(msg.Key, check.BytesEquals, []byte("key1"))
 	c.Assert(msg.Value, check.BytesEquals, []byte("value1"))
 	c.Assert(msg.Ts, check.Equals, rowEvent.CommitTs)
 	c.Assert(msg.Type, check.Equals, model.MqMessageTypeRow)
 	c.Assert(*msg.Schema, check.Equals, rowEvent.Table.Schema)
 	c.Assert(*msg.Table, check.Equals, rowEvent.Table.Table)
-	c.Assert(msg.Protocol, check.Equals, ProtocolDefault)
+	c.Assert(msg.Protocol, check.Equals, config.ProtocolDefault)
 
 	job := &timodel.Job{
 		ID:         1071,
@@ -100,21 +101,21 @@ func (s *codecInterfaceSuite) TestCreate(c *check.C) {
 	ddlEvent := &model.DDLEvent{}
 	ddlEvent.FromJob(job, preTableInfo)
 
-	msg = newDDLMQMessage(ProtocolMaxwell, nil, []byte("value1"), ddlEvent)
+	msg = newDDLMQMessage(config.ProtocolMaxwell, nil, []byte("value1"), ddlEvent)
 	c.Assert(msg.Key, check.IsNil)
 	c.Assert(msg.Value, check.BytesEquals, []byte("value1"))
 	c.Assert(msg.Ts, check.Equals, ddlEvent.CommitTs)
 	c.Assert(msg.Type, check.Equals, model.MqMessageTypeDDL)
 	c.Assert(*msg.Schema, check.Equals, ddlEvent.TableInfo.Schema)
 	c.Assert(*msg.Table, check.Equals, ddlEvent.TableInfo.Table)
-	c.Assert(msg.Protocol, check.Equals, ProtocolMaxwell)
+	c.Assert(msg.Protocol, check.Equals, config.ProtocolMaxwell)
 
-	msg = newResolvedMQMessage(ProtocolCanal, []byte("key1"), nil, 1234)
+	msg = newResolvedMQMessage(config.ProtocolCanal, []byte("key1"), nil, 1234)
 	c.Assert(msg.Key, check.BytesEquals, []byte("key1"))
 	c.Assert(msg.Value, check.IsNil)
 	c.Assert(msg.Ts, check.Equals, uint64(1234))
 	c.Assert(msg.Type, check.Equals, model.MqMessageTypeResolved)
 	c.Assert(msg.Schema, check.IsNil)
 	c.Assert(msg.Table, check.IsNil)
-	c.Assert(msg.Protocol, check.Equals, ProtocolCanal)
+	c.Assert(msg.Protocol, check.Equals, config.ProtocolCanal)
 }
diff --git a/cdc/sink/codec/json.go b/cdc/sink/codec/json.go
@@ -25,12 +25,12 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"github.com/pingcap/ticdc/cdc/model"
+	"github.com/pingcap/ticdc/pkg/config"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
 	timodel "github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	"go.uber.org/zap"
-
-	"github.com/pingcap/ticdc/cdc/model"
 )
 
 const (
@@ -398,7 +398,7 @@ func (d *JSONEventBatchEncoder) EncodeCheckpointEvent(ts uint64) (*MQMessage, er
 	valueBuf := new(bytes.Buffer)
 	valueBuf.Write(valueLenByte[:])
 
-	ret := newResolvedMQMessage(ProtocolDefault, keyBuf.Bytes(), valueBuf.Bytes(), ts)
+	ret := newResolvedMQMessage(config.ProtocolDefault, keyBuf.Bytes(), valueBuf.Bytes(), ts)
 	return ret, nil
 }
 
@@ -442,7 +442,7 @@ func (d *JSONEventBatchEncoder) AppendRowChangedEvent(e *model.RowChangedEvent)
 			versionHead := make([]byte, 8)
 			binary.BigEndian.PutUint64(versionHead, BatchVersion1)
 
-			d.messageBuf = append(d.messageBuf, NewMQMessage(ProtocolDefault, versionHead, nil, 0, model.MqMessageTypeRow, nil, nil))
+			d.messageBuf = append(d.messageBuf, NewMQMessage(config.ProtocolDefault, versionHead, nil, 0, model.MqMessageTypeRow, nil, nil))
 			d.curBatchSize = 0
 		}
 
@@ -501,7 +501,7 @@ func (d *JSONEventBatchEncoder) EncodeDDLEvent(e *model.DDLEvent) (*MQMessage, e
 	valueBuf.Write(valueLenByte[:])
 	valueBuf.Write(value)
 
-	ret := newDDLMQMessage(ProtocolDefault, keyBuf.Bytes(), valueBuf.Bytes(), e)
+	ret := newDDLMQMessage(config.ProtocolDefault, keyBuf.Bytes(), valueBuf.Bytes(), e)
 	return ret, nil
 }
 
@@ -512,7 +512,7 @@ func (d *JSONEventBatchEncoder) Build() (mqMessages []*MQMessage) {
 			return nil
 		}
 		/* there could be multiple types of event encoded within a single message which means the type is not sure */
-		ret := NewMQMessage(ProtocolDefault, d.keyBuf.Bytes(), d.valueBuf.Bytes(), 0, model.MqMessageTypeUnknown, nil, nil)
+		ret := NewMQMessage(config.ProtocolDefault, d.keyBuf.Bytes(), d.valueBuf.Bytes(), 0, model.MqMessageTypeUnknown, nil, nil)
 		return []*MQMessage{ret}
 	}
 

diff --git a/cdc/sink/codec/maxwell.go b/cdc/sink/codec/maxwell.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/ticdc/cdc/model"
+	"github.com/pingcap/ticdc/pkg/config"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
 	model2 "github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
@@ -287,7 +288,7 @@ func (d *MaxwellEventBatchEncoder) EncodeDDLEvent(e *model.DDLEvent) (*MQMessage
 		return nil, errors.Trace(err)
 	}
 
-	return newDDLMQMessage(ProtocolMaxwell, key, value, e), nil
+	return newDDLMQMessage(config.ProtocolMaxwell, key, value, e), nil
 }
 
 // Build implements the EventBatchEncoder interface
@@ -296,7 +297,7 @@ func (d *MaxwellEventBatchEncoder) Build() []*MQMessage {
 		return nil
 	}
 
-	ret := NewMQMessage(ProtocolMaxwell, d.keyBuf.Bytes(), d.valueBuf.Bytes(), 0, model.MqMessageTypeRow, nil, nil)
+	ret := NewMQMessage(config.ProtocolMaxwell, d.keyBuf.Bytes(), d.valueBuf.Bytes(), 0, model.MqMessageTypeRow, nil, nil)
 	d.Reset()
 	return []*MQMessage{ret}
 }