codec(ticdc): simple support large message handle functionality

cdc/model/changefeed.go

@@ -571,42 +571,6 @@ func (info *ChangeFeedInfo) updateSinkURIAndConfigProtocol(uri *url.URL, newProt
 	info.Config.Sink.Protocol = util.AddressOf(newProtocol)
 }
 
-// DownstreamType returns the type of the downstream.
-func (info *ChangeFeedInfo) DownstreamType() (DownstreamType, error) {
-	uri, err := url.Parse(info.SinkURI)
-	if err != nil {
-		return Unknown, errors.Trace(err)
-	}
-	if sink.IsMySQLCompatibleScheme(uri.Scheme) {
-		return DB, nil
-	}
-	if sink.IsMQScheme(uri.Scheme) {
-		return MQ, nil
-	}
-	if sink.IsStorageScheme(uri.Scheme) {
-		return Storage, nil
-	}
-	return Unknown, nil
-}
-
-// NeedSendBootstrapEvent returns true if the changefeed need to send bootstrap event.
-func (info *ChangeFeedInfo) NeedSendBootstrapEvent() (bool, error) {
-	downStreamType, err := info.DownstreamType()
-	if err != nil {
-		return false, errors.Trace(err)
-	}
-	if downStreamType != MQ {
-		return false, nil
-	}
-	if info.Config.Sink.Protocol == nil {
-		return false, nil
-	}
-	if *info.Config.Sink.Protocol == config.ProtocolSimple.String() {
-		return true, nil
-	}
-	return false, nil
-}
-
 func (info *ChangeFeedInfo) fixMemoryQuota() {
 	info.Config.FixMemoryQuota()
 }
@@ -615,33 +579,6 @@ func (info *ChangeFeedInfo) fixScheduler(inheritV66 bool) {
 	info.Config.FixScheduler(inheritV66)
 }
 
-// DownstreamType is the type of downstream.
-type DownstreamType int
-
-const (
-	// DB is the type of Database.
-	DB DownstreamType = iota
-	// MQ is the type of MQ or Cloud Storage.
-	MQ
-	// Storage is the type of Cloud Storage.
-	Storage
-	// Unknown is the type of Unknown.
-	Unknown
-)
-
-// String implements fmt.Stringer interface.
-func (t DownstreamType) String() string {
-	switch t {
-	case DB:
-		return "DB"
-	case MQ:
-		return "MQ"
-	case Storage:
-		return "Storage"
-	}
-	return "Unknown"
-}
-
 // ChangeFeedStatusForAPI uses to transfer the status of changefeed for API.
 type ChangeFeedStatusForAPI struct {
 	ResolvedTs   uint64 `json:"resolved-ts"`

cdc/owner/changefeed.go

@@ -634,16 +634,6 @@ LOOP2:
 		zap.String("namespace", c.id.Namespace),
 		zap.String("changefeed", c.id.ID))
 
-	downstreamType, err := c.latestInfo.DownstreamType()
-	if err != nil {
-		return errors.Trace(err)
-	}
-
-	needSendBootstrapEvent, err := c.latestInfo.NeedSendBootstrapEvent()
-	if err != nil {
-		return errors.Trace(err)
-	}
-
 	c.ddlManager = newDDLManager(
 		c.id,
 		ddlStartTs,
@@ -653,10 +643,7 @@ LOOP2:
 		c.schema,
 		c.redoDDLMgr,
 		c.redoMetaMgr,
-		downstreamType,
-		util.GetOrZero(c.latestInfo.Config.BDRMode),
-		needSendBootstrapEvent,
-	)
+		util.GetOrZero(c.latestInfo.Config.BDRMode))
 
 	// create scheduler
 	cfg := *c.cfg

cdc/owner/changefeed_test.go

@@ -140,10 +140,6 @@ func (m *mockDDLSink) emitCheckpointTs(ts uint64, tables []*model.TableInfo) {
 	m.mu.currentTables = tables
 }
 
-func (m *mockDDLSink) emitBootstrapEvent(ctx context.Context, ddl *model.DDLEvent) error {
-	return nil
-}
-
 func (m *mockDDLSink) getCheckpointTsAndTableNames() (uint64, []*model.TableInfo) {
 	m.mu.Lock()
 	defer m.mu.Unlock()

cdc/owner/ddl_manager.go

@@ -17,7 +17,6 @@ import (
 	"context"
 	"math/rand"
 	"sort"
-	"sync/atomic"
 	"time"
 
 	"github.com/pingcap/errors"
@@ -94,14 +93,6 @@ var redoBarrierDDLs = map[timodel.ActionType]struct{}{
 	timodel.ActionRemovePartitioning:     {},
 }
 
-type bootstrapState int32
-
-const (
-	bootstrapStateNone bootstrapState = iota
-	bootstrapStateRunning
-	bootstrapStateCompleted
-)
-
 // ddlManager holds the pending DDL events of all tables and responsible for
 // executing them to downstream.
 // It also provides the ability to calculate the barrier of a changefeed.
@@ -136,12 +127,6 @@ type ddlManager struct {
 
 	BDRMode       bool
 	ddlResolvedTs model.Ts
-
-	// needBootstrap is true when the downstream is kafka
-	// and the protocol is simple protocol.
-	needSendBootstrapEvent bool
-	errCh                  chan error
-	bootstrapState         int32
 }
 
 func newDDLManager(
@@ -153,32 +138,27 @@ func newDDLManager(
 	schema *schemaWrap4Owner,
 	redoManager redo.DDLManager,
 	redoMetaManager redo.MetaManager,
-	sinkType model.DownstreamType,
 	bdrMode bool,
-	needSendBootstrapEvent bool,
 ) *ddlManager {
 	log.Info("owner create ddl manager",
 		zap.String("namespace", changefeedID.Namespace),
 		zap.String("changefeed", changefeedID.ID),
 		zap.Uint64("startTs", startTs),
 		zap.Uint64("checkpointTs", checkpointTs),
-		zap.Bool("bdrMode", bdrMode),
-		zap.Stringer("sinkType", sinkType))
+		zap.Bool("bdrMode", bdrMode))
 
 	return &ddlManager{
-		changfeedID:            changefeedID,
-		ddlSink:                ddlSink,
-		ddlPuller:              ddlPuller,
-		schema:                 schema,
-		redoDDLManager:         redoManager,
-		redoMetaManager:        redoMetaManager,
-		startTs:                startTs,
-		checkpointTs:           checkpointTs,
-		ddlResolvedTs:          startTs,
-		BDRMode:                bdrMode,
-		pendingDDLs:            make(map[model.TableName][]*model.DDLEvent),
-		errCh:                  make(chan error, 1),
-		needSendBootstrapEvent: needSendBootstrapEvent,
+		changfeedID:     changefeedID,
+		ddlSink:         ddlSink,
+		ddlPuller:       ddlPuller,
+		schema:          schema,
+		redoDDLManager:  redoManager,
+		redoMetaManager: redoMetaManager,
+		startTs:         startTs,
+		checkpointTs:    checkpointTs,
+		ddlResolvedTs:   startTs,
+		BDRMode:         bdrMode,
+		pendingDDLs:     make(map[model.TableName][]*model.DDLEvent),
 	}
 }
 
@@ -195,16 +175,6 @@ func (m *ddlManager) tick(
 	ctx context.Context,
 	checkpointTs model.Ts,
 ) ([]model.TableID, *schedulepb.BarrierWithMinTs, error) {
-	if m.needSendBootstrapEvent {
-		finished, err := m.checkAndBootstrap(ctx)
-		if err != nil {
-			return nil, nil, err
-		}
-		if !finished {
-			return nil, schedulepb.NewBarrierWithMinTs(checkpointTs), nil
-		}
-	}
-
 	m.justSentDDL = nil
 	m.checkpointTs = checkpointTs
 
@@ -594,53 +564,6 @@ func (m *ddlManager) cleanCache() {
 	m.physicalTablesCache = nil
 }
 
-func (m *ddlManager) checkAndBootstrap(ctx context.Context) (bool, error) {
-	if atomic.LoadInt32(&m.bootstrapState) == int32(bootstrapStateCompleted) {
-		return true, nil
-	}
-
-	select {
-	case err := <-m.errCh:
-		return false, err
-	default:
-	}
-
-	if atomic.LoadInt32(&m.bootstrapState) == int32(bootstrapStateRunning) {
-		return false, nil
-	}
-	// begin bootstrap
-	atomic.StoreInt32(&m.bootstrapState, int32(bootstrapStateRunning))
-	tables, err := m.allTables(ctx)
-	if err != nil {
-		return false, err
-	}
-	bootstrapEvents := make([]*model.DDLEvent, 0, len(tables))
-	for _, table := range tables {
-		ddlEvent := &model.DDLEvent{
-			StartTs:     m.startTs,
-			CommitTs:    m.startTs,
-			TableInfo:   table,
-			IsBootstrap: true,
-		}
-		bootstrapEvents = append(bootstrapEvents, ddlEvent)
-	}
-	// send bootstrap events
-	go func() {
-		for _, event := range bootstrapEvents {
-			err := m.ddlSink.emitBootstrapEvent(ctx, event)
-			if err != nil {
-				log.Error("emit bootstrap event failed",
-					zap.Any("bootstrapEvent", event), zap.Error(err))
-				atomic.StoreInt32(&m.bootstrapState, int32(bootstrapStateNone))
-				m.errCh <- err
-				return
-			}
-		}
-		atomic.StoreInt32(&m.bootstrapState, int32(bootstrapStateCompleted))
-	}()
-	return false, nil
-}
-
 // getRelatedPhysicalTableIDs get all related physical table ids of a ddl event.
 // It is a helper function to calculate tableBarrier.
 func getRelatedPhysicalTableIDs(ddl *model.DDLEvent) []model.TableID {

cdc/owner/ddl_manager_test.go

@@ -48,7 +48,7 @@ func createDDLManagerForTest(t *testing.T) *ddlManager {
 		schema,
 		redo.NewDisabledDDLManager(),
 		redo.NewDisabledMetaManager(),
-		model.DB, false, false)
+		false)
 	return res
 }
 

cdc/owner/ddl_sink.go

@@ -54,7 +54,6 @@ type DDLSink interface {
 	// the DDL event will be sent to another goroutine and execute to downstream
 	// the caller of this function can call again and again until a true returned
 	emitDDLEvent(ctx context.Context, ddl *model.DDLEvent) (bool, error)
-	emitBootstrapEvent(ctx context.Context, ddl *model.DDLEvent) error
 	emitSyncPoint(ctx context.Context, checkpointTs uint64) error
 	// close the ddlsink, cancel running goroutine.
 	close(ctx context.Context) error
@@ -385,22 +384,6 @@ func (s *ddlSinkImpl) emitDDLEvent(ctx context.Context, ddl *model.DDLEvent) (bo
 	return false, nil
 }
 
-// emitBootstrapEvent sent bootstrap event to downstream.
-// It is a synchronous operation.
-func (s *ddlSinkImpl) emitBootstrapEvent(ctx context.Context, ddl *model.DDLEvent) error {
-	if !ddl.IsBootstrap {
-		return nil
-	}
-	err := s.sink.WriteDDLEvent(ctx, ddl)
-	if err != nil {
-		return errors.Trace(err)
-	}
-	// TODO: change this log to debug level after testing complete.
-	log.Info("emit bootstrap event", zap.String("namespace", s.changefeedID.Namespace),
-		zap.String("changefeed", s.changefeedID.ID), zap.Any("bootstrapEvent", ddl))
-	return nil
-}
-
 func (s *ddlSinkImpl) emitSyncPoint(ctx context.Context, checkpointTs uint64) (err error) {
 	if checkpointTs == s.lastSyncPoint {
 		return nil

cmd/kafka-consumer/main.go

@@ -630,7 +630,7 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 		}
 		decoder = avro.NewDecoder(c.codecConfig, schemaM, c.option.topic, c.tz)
 	case config.ProtocolSimple:
-		decoder = simple.NewDecoder()
+		decoder, err = simple.NewDecoder(ctx, c.codecConfig, c.upstreamTiDB)
 	default:
 		log.Panic("Protocol not supported", zap.Any("Protocol", c.codecConfig.Protocol))
 	}

pkg/config/large_message.go

@@ -62,7 +62,7 @@ func (c *LargeMessageHandleConfig) AdjustAndValidate(protocol Protocol, enableTi
 	}
 
 	switch protocol {
-	case ProtocolOpen:
+	case ProtocolOpen, ProtocolSimple:
 	case ProtocolCanalJSON:
 		if !enableTiDBExtension {
 			return cerror.ErrInvalidReplicaConfig.GenWithStack(

pkg/sink/codec/builder/encoder_builder.go

@@ -51,7 +51,7 @@ func NewRowEventEncoderBuilder(
 	case config.ProtocolDebezium:
 		return debezium.NewBatchEncoderBuilder(cfg), nil
 	case config.ProtocolSimple:
-		return simple.NewBuilder(cfg), nil
+		return simple.NewBuilder(ctx, cfg)
 	default:
 		return nil, cerror.ErrSinkUnknownProtocol.GenWithStackByArgs(cfg.Protocol)
 	}

pkg/sink/codec/canal/canal_json_row_event_encoder_test.go

@@ -41,7 +41,7 @@ func TestBuildCanalJSONRowEventEncoder(t *testing.T) {
 }
 
 func TestDMLE2E(t *testing.T) {
-	insertEvent, updateEvent, deleteEvent := utils.NewLargeEvent4Test(t)
+	_, insertEvent, updateEvent, deleteEvent := utils.NewLargeEvent4Test(t)
 
 	ctx := context.Background()
 
@@ -126,7 +126,7 @@ func TestDMLE2E(t *testing.T) {
 }
 
 func TestCanalJSONCompressionE2E(t *testing.T) {
-	insertEvent, _, _ := utils.NewLargeEvent4Test(t)
+	_, insertEvent, _, _ := utils.NewLargeEvent4Test(t)
 
 	codecConfig := common.NewConfig(config.ProtocolCanalJSON)
 	codecConfig.EnableTiDBExtension = true
@@ -217,7 +217,7 @@ func TestCanalJSONClaimCheckE2E(t *testing.T) {
 	require.NoError(t, err)
 	encoder := builder.Build()
 
-	insertEvent, _, _ := utils.NewLargeEvent4Test(t)
+	_, insertEvent, _, _ := utils.NewLargeEvent4Test(t)
 	err = encoder.AppendRowChangedEvent(ctx, "", insertEvent, func() {})
 	require.NoError(t, err)
 
@@ -269,7 +269,7 @@ func TestNewCanalJSONMessageHandleKeyOnly4LargeMessage(t *testing.T) {
 	require.NoError(t, err)
 	encoder := builder.Build()
 
-	insertEvent, _, _ := utils.NewLargeEvent4Test(t)
+	_, insertEvent, _, _ := utils.NewLargeEvent4Test(t)
 	err = encoder.AppendRowChangedEvent(context.Background(), "", insertEvent, func() {})
 	require.NoError(t, err)
 
@@ -351,7 +351,7 @@ func TestBatching(t *testing.T) {
 	encoder := builder.Build()
 	require.NotNil(t, encoder)
 
-	_, updateEvent, _ := utils.NewLargeEvent4Test(t)
+	_, _, updateEvent, _ := utils.NewLargeEvent4Test(t)
 	updateCase := *updateEvent
 	for i := 1; i <= 1000; i++ {
 		ts := uint64(i)
@@ -629,7 +629,7 @@ func TestCanalJSONContentCompatibleE2E(t *testing.T) {
 
 	encoder := builder.Build()
 
-	insertEvent, _, _ := utils.NewLargeEvent4Test(t)
+	_, insertEvent, _, _ := utils.NewLargeEvent4Test(t)
 	err = encoder.AppendRowChangedEvent(ctx, "", insertEvent, func() {})
 	require.NoError(t, err)
 
@@ -666,7 +666,7 @@ func TestCanalJSONContentCompatibleE2E(t *testing.T) {
 }
 
 func TestNewCanalJSONBatchDecoder4RowMessage(t *testing.T) {
-	insertEvent, _, _ := utils.NewLargeEvent4Test(t)
+	_, insertEvent, _, _ := utils.NewLargeEvent4Test(t)
 	ctx := context.Background()
 
 	for _, encodeEnable := range []bool{false, true} {