Merge branch 'master' into debezium

Author: wk989898

downstreamadapter/dispatcher/dispatcher.go

@@ -235,7 +235,7 @@ func (d *Dispatcher) HandleDispatcherStatus(dispatcherStatus *heartbeatpb.Dispat
 	// deal with the dispatcher action
 	action := dispatcherStatus.GetAction()
 	if action != nil {
-		pendingEvent, blockStatus := d.blockEventStatus.getEventAndStage()
+		pendingEvent := d.blockEventStatus.getEvent()
 		if pendingEvent == nil && action.CommitTs > d.GetResolvedTs() {
 			// we have not received the block event, and the action is for the future event, so just ignore
 			log.Info("pending event is nil, and the action's commit is larger than dispatchers resolvedTs",
@@ -245,7 +245,7 @@ func (d *Dispatcher) HandleDispatcherStatus(dispatcherStatus *heartbeatpb.Dispat
 			// we have not received the block event, and the action is for the future event, so just ignore
 			return
 		}
-		if pendingEvent != nil && action.CommitTs == pendingEvent.GetCommitTs() && blockStatus == heartbeatpb.BlockStage_WAITING {
+		if d.blockEventStatus.actionMatchs(action) {
 			log.Info("pending event get the action",
 				zap.Any("action", action),
 				zap.Stringer("dispatcher", d.id),
@@ -404,8 +404,9 @@ func (d *Dispatcher) HandleEvents(dispatcherEvents []DispatcherEvent, wakeCallba
 			syncPoint := event.(*commonEvent.SyncPointEvent)
 			log.Info("dispatcher receive sync point event",
 				zap.Stringer("dispatcher", d.id),
-				zap.Uint64("commitTs", event.GetCommitTs()),
+				zap.Any("commitTsList", syncPoint.GetCommitTsList()),
 				zap.Uint64("seq", event.GetSeq()))
+
 			syncPoint.AddPostFlushFunc(func() {
 				wakeCallback()
 			})
@@ -546,25 +547,57 @@ func (d *Dispatcher) dealWithBlockEvent(event commonEvent.BlockEvent) {
 		}
 	} else {
 		d.blockEventStatus.setBlockEvent(event, heartbeatpb.BlockStage_WAITING)
-		message := &heartbeatpb.TableSpanBlockStatus{
-			ID: d.id.ToPB(),
-			State: &heartbeatpb.State{
-				IsBlocked:         true,
-				BlockTs:           event.GetCommitTs(),
-				BlockTables:       event.GetBlockedTables().ToPB(),
-				NeedDroppedTables: event.GetNeedDroppedTables().ToPB(),
-				NeedAddedTables:   commonEvent.ToTablesPB(event.GetNeedAddedTables()),
-				UpdatedSchemas:    commonEvent.ToSchemaIDChangePB(event.GetUpdatedSchemas()), // only exists for rename table and rename tables
-				IsSyncPoint:       event.GetType() == commonEvent.TypeSyncPointEvent,         // sync point event must should block
-				Stage:             heartbeatpb.BlockStage_WAITING,
-			},
-		}
-		identifier := BlockEventIdentifier{
-			CommitTs:    event.GetCommitTs(),
-			IsSyncPoint: event.GetType() == commonEvent.TypeSyncPointEvent,
+
+		if event.GetType() == commonEvent.TypeSyncPointEvent {
+			// deal with multi sync point commit ts in one Sync Point Event
+			// make each commitTs as a single message for maintainer
+			// Because the batch commitTs in different dispatchers can be different.
+			commitTsList := event.(*commonEvent.SyncPointEvent).GetCommitTsList()
+			blockTables := event.GetBlockedTables().ToPB()
+			needDroppedTables := event.GetNeedDroppedTables().ToPB()
+			needAddedTables := commonEvent.ToTablesPB(event.GetNeedAddedTables())
+			for _, commitTs := range commitTsList {
+				message := &heartbeatpb.TableSpanBlockStatus{
+					ID: d.id.ToPB(),
+					State: &heartbeatpb.State{
+						IsBlocked:         true,
+						BlockTs:           commitTs,
+						BlockTables:       blockTables,
+						NeedDroppedTables: needDroppedTables,
+						NeedAddedTables:   needAddedTables,
+						UpdatedSchemas:    nil,
+						IsSyncPoint:       true,
+						Stage:             heartbeatpb.BlockStage_WAITING,
+					},
+				}
+				identifier := BlockEventIdentifier{
+					CommitTs:    commitTs,
+					IsSyncPoint: true,
+				}
+				d.resendTaskMap.Set(identifier, newResendTask(message, d, nil))
+				d.blockStatusesChan <- message
+			}
+		} else {
+			message := &heartbeatpb.TableSpanBlockStatus{
+				ID: d.id.ToPB(),
+				State: &heartbeatpb.State{
+					IsBlocked:         true,
+					BlockTs:           event.GetCommitTs(),
+					BlockTables:       event.GetBlockedTables().ToPB(),
+					NeedDroppedTables: event.GetNeedDroppedTables().ToPB(),
+					NeedAddedTables:   commonEvent.ToTablesPB(event.GetNeedAddedTables()),
+					UpdatedSchemas:    commonEvent.ToSchemaIDChangePB(event.GetUpdatedSchemas()), // only exists for rename table and rename tables
+					IsSyncPoint:       false,
+					Stage:             heartbeatpb.BlockStage_WAITING,
+				},
+			}
+			identifier := BlockEventIdentifier{
+				CommitTs:    event.GetCommitTs(),
+				IsSyncPoint: false,
+			}
+			d.resendTaskMap.Set(identifier, newResendTask(message, d, nil))
+			d.blockStatusesChan <- message
 		}
-		d.resendTaskMap.Set(identifier, newResendTask(message, d, nil))
-		d.blockStatusesChan <- message
 	}
 
 	// dealing with events which update schema ids

downstreamadapter/dispatcher/dispatcher_test.go

@@ -363,7 +363,7 @@ func TestDispatcherHandleEvents(t *testing.T) {
 	// ===== sync point event =====
 
 	syncPointEvent := &commonEvent.SyncPointEvent{
-		CommitTs: 6,
+		CommitTsList: []uint64{6},
 	}
 	block = dispatcher.HandleEvents([]DispatcherEvent{NewDispatcherEvent(&nodeID, syncPointEvent)}, callback)
 	require.Equal(t, true, block)
@@ -380,7 +380,7 @@ func TestDispatcherHandleEvents(t *testing.T) {
 	// receive the ack info
 	dispatcherStatus = &heartbeatpb.DispatcherStatus{
 		Ack: &heartbeatpb.ACK{
-			CommitTs:    syncPointEvent.CommitTs,
+			CommitTs:    syncPointEvent.GetCommitTs(),
 			IsSyncPoint: true,
 		},
 	}
@@ -394,7 +394,7 @@ func TestDispatcherHandleEvents(t *testing.T) {
 	dispatcherStatus = &heartbeatpb.DispatcherStatus{
 		Action: &heartbeatpb.DispatcherAction{
 			Action:      heartbeatpb.Action_Pass,
-			CommitTs:    syncPointEvent.CommitTs,
+			CommitTs:    syncPointEvent.GetCommitTs(),
 			IsSyncPoint: true,
 		},
 	}

downstreamadapter/dispatcher/helper.go

@@ -74,6 +74,9 @@ type BlockEventStatus struct {
 	mutex             sync.Mutex
 	blockPendingEvent commonEvent.BlockEvent
 	blockStage        heartbeatpb.BlockStage
+	// record all the commitTs of this pending event
+	// mainly for the batch sync point event
+	blockCommitTsMap map[uint64]struct{}
 }
 
 func (b *BlockEventStatus) clear() {
@@ -82,6 +85,7 @@ func (b *BlockEventStatus) clear() {
 
 	b.blockPendingEvent = nil
 	b.blockStage = heartbeatpb.BlockStage_NONE
+	b.blockCommitTsMap = make(map[uint64]struct{})
 }
 
 func (b *BlockEventStatus) setBlockEvent(event commonEvent.BlockEvent, blockStage heartbeatpb.BlockStage) {
@@ -90,6 +94,15 @@ func (b *BlockEventStatus) setBlockEvent(event commonEvent.BlockEvent, blockStag
 
 	b.blockPendingEvent = event
 	b.blockStage = blockStage
+	b.blockCommitTsMap = make(map[uint64]struct{})
+
+	if event.GetType() == commonEvent.TypeSyncPointEvent {
+		for _, ts := range event.(*commonEvent.SyncPointEvent).GetCommitTsList() {
+			b.blockCommitTsMap[ts] = struct{}{}
+		}
+	} else {
+		b.blockCommitTsMap[event.GetCommitTs()] = struct{}{}
+	}
 }
 
 func (b *BlockEventStatus) updateBlockStage(blockStage heartbeatpb.BlockStage) {
@@ -98,13 +111,47 @@ func (b *BlockEventStatus) updateBlockStage(blockStage heartbeatpb.BlockStage) {
 	b.blockStage = blockStage
 }
 
+func (b *BlockEventStatus) getEvent() commonEvent.BlockEvent {
+	b.mutex.Lock()
+	defer b.mutex.Unlock()
+
+	return b.blockPendingEvent
+}
+
 func (b *BlockEventStatus) getEventAndStage() (commonEvent.BlockEvent, heartbeatpb.BlockStage) {
 	b.mutex.Lock()
 	defer b.mutex.Unlock()
 
 	return b.blockPendingEvent, b.blockStage
 }
 
+// actionMatchs checks whether the action is for the current pending ddl event.
+// Most of time, the pending event only have one commitTs, so when the commitTs of the action meets the pending event's commitTs, it is enough.
+// While if the pending event is a sync point event with multiple commitTs, we only can do the action
+// when all the commitTs have been received.
+func (b *BlockEventStatus) actionMatchs(action *heartbeatpb.DispatcherAction) bool {
+	b.mutex.Lock()
+	defer b.mutex.Unlock()
+
+	if b.blockPendingEvent == nil {
+		return false
+	}
+
+	if b.blockStage != heartbeatpb.BlockStage_WAITING {
+		return false
+	}
+
+	_, ok := b.blockCommitTsMap[action.CommitTs]
+	if ok {
+		delete(b.blockCommitTsMap, action.CommitTs)
+	}
+
+	if len(b.blockCommitTsMap) == 0 {
+		return true
+	}
+	return false
+}
+
 type SchemaIDToDispatchers struct {
 	mutex sync.RWMutex
 	m     map[int64]map[common.DispatcherID]interface{}

downstreamadapter/dispatcher/table_progress.go

@@ -62,12 +62,21 @@ func NewTableProgress() *TableProgress {
 
 // Add inserts a new event into the TableProgress.
 func (p *TableProgress) Add(event commonEvent.FlushableEvent) {
-	ts := Ts{startTs: event.GetStartTs(), commitTs: event.GetCommitTs()}
+	commitTs := event.GetCommitTs()
+	if event.GetType() == commonEvent.TypeSyncPointEvent {
+		// if the event is a sync point event, we use the last commitTs(the largest commitTs in the event) to calculate the progress.
+		// because a sync point event with multiple commitTs means there is no ddl / dmls between these commitTses.
+		// So we can just use the largest commitTs in the sync point event to calculate the progress.
+		commitTsList := event.(*commonEvent.SyncPointEvent).GetCommitTsList()
+		commitTs = commitTsList[len(commitTsList)-1]
+	}
+
+	ts := Ts{startTs: event.GetStartTs(), commitTs: commitTs}
 	p.rwMutex.Lock()
 	defer p.rwMutex.Unlock()
 	elem := p.list.PushBack(ts)
 	p.elemMap[ts] = elem
-	p.maxCommitTs = event.GetCommitTs()
+	p.maxCommitTs = commitTs
 	event.PushFrontFlushFunc(func() {
 		p.Remove(event)
 	})
@@ -98,7 +107,16 @@ func (p *TableProgress) Empty() bool {
 func (p *TableProgress) Pass(event commonEvent.BlockEvent) {
 	p.rwMutex.Lock()
 	defer p.rwMutex.Unlock()
+
 	p.maxCommitTs = event.GetCommitTs()
+
+	if event.GetType() == commonEvent.TypeSyncPointEvent {
+		// if the event is a sync point event, we use the last commitTs(the largest commitTs in the event) to calculate the progress.
+		// because a sync point event with multiple commitTs means there is no ddl / dmls between these commitTses.
+		// So we can just use the largest commitTs in the sync point event to calculate the progress.
+		commitTsList := event.(*commonEvent.SyncPointEvent).GetCommitTsList()
+		p.maxCommitTs = commitTsList[len(commitTsList)-1]
+	}
 }
 
 // GetCheckpointTs returns the current checkpoint timestamp for the table span.

downstreamadapter/dispatcherorchestrator/dispatcher_orchestrator.go

@@ -16,6 +16,7 @@ package dispatcherorchestrator
 import (
 	"context"
 	"encoding/json"
+	"sync"
 	"time"
 
 	"github.com/pingcap/log"
@@ -36,6 +37,7 @@ import (
 // for different change feeds based on maintainer bootstrap messages.
 type DispatcherOrchestrator struct {
 	mc                 messaging.MessageCenter
+	mutex              sync.Mutex // protect dispatcherManagers
 	dispatcherManagers map[common.ChangeFeedID]*dispatchermanager.EventDispatcherManager
 }
 
@@ -81,7 +83,10 @@ func (m *DispatcherOrchestrator) handleBootstrapRequest(
 		return err
 	}
 
+	m.mutex.Lock()
 	manager, exists := m.dispatcherManagers[cfId]
+	m.mutex.Unlock()
+
 	var err error
 	var startTs uint64
 	if !exists {
@@ -112,7 +117,9 @@ func (m *DispatcherOrchestrator) handleBootstrapRequest(
 			}
 			return m.sendResponse(from, messaging.MaintainerManagerTopic, response)
 		}
+		m.mutex.Lock()
 		m.dispatcherManagers[cfId] = manager
+		m.mutex.Unlock()
 		metrics.EventDispatcherManagerGauge.WithLabelValues(cfId.Namespace(), cfId.Name()).Inc()
 	} else {
 		// Check and potentially add a table trigger event dispatcher.
@@ -164,7 +171,11 @@ func (m *DispatcherOrchestrator) handlePostBootstrapRequest(
 	req *heartbeatpb.MaintainerPostBootstrapRequest,
 ) error {
 	cfId := common.NewChangefeedIDFromPB(req.ChangefeedID)
+
+	m.mutex.Lock()
 	manager, exists := m.dispatcherManagers[cfId]
+	m.mutex.Unlock()
+
 	if !exists || manager.GetTableTriggerEventDispatcher() == nil {
 		log.Error("Receive post bootstrap request but there is no table trigger event dispatcher",
 			zap.Any("changefeedID", cfId.Name()))
@@ -220,6 +231,7 @@ func (m *DispatcherOrchestrator) handleCloseRequest(
 		Success:      true,
 	}
 
+	m.mutex.Lock()
 	if manager, ok := m.dispatcherManagers[cfId]; ok {
 		if closed := manager.TryClose(req.Removed); closed {
 			delete(m.dispatcherManagers, cfId)
@@ -229,6 +241,7 @@ func (m *DispatcherOrchestrator) handleCloseRequest(
 			response.Success = false
 		}
 	}
+	m.mutex.Unlock()
 
 	log.Info("try close dispatcher manager",
 		zap.String("changefeed", cfId.Name()), zap.Bool("success", response.Success))
@@ -275,6 +288,17 @@ func (m *DispatcherOrchestrator) sendResponse(to node.ID, topic string, msg mess
 func (m *DispatcherOrchestrator) Close() {
 	log.Info("dispatcher orchestrator is closing")
 	m.mc.DeRegisterHandler(messaging.DispatcherManagerManagerTopic)
+
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	for len(m.dispatcherManagers) > 0 {
+		for id, manager := range m.dispatcherManagers {
+			ok := manager.TryClose(false)
+			if ok {
+				delete(m.dispatcherManagers, id)
+			}
+		}
+	}
 	log.Info("dispatcher orchestrator closed")
 }