fix race

downstreamadapter/dispatcher/dispatcher.go

@@ -235,9 +235,11 @@ func (d *Dispatcher) HandleEvents(dispatcherEvents []DispatcherEvent) (block boo
 			log.Info("Received a stale event, ignore it", zap.Any("event", event), zap.Any("dispatcher", d.id))
 		}
 		if event.GetType() == commonEvent.TypeDMLEvent ||
-			event.GetType() == commonEvent.TypeDDLEvent {
+			event.GetType() == commonEvent.TypeDDLEvent ||
+			event.GetType() == commonEvent.TypeHandshakeEvent {
 			if event.GetSeq() != d.lastEventSeq.Add(1) {
-				log.Warn("Received a out-of-order event, reset the dispatcher", zap.Any("event", event), zap.Any("dispatcher", d.id))
+				log.Warn("Received a out-of-order event, reset the dispatcher", zap.Any("dispatcher", d.id),
+					zap.Uint64("receivedSeq", event.GetSeq()), zap.Uint64("lastEventSeq", d.lastEventSeq.Load()), zap.Any("event", event))
 				d.reset()
 				return false
 			}

downstreamadapter/dispatcher/helper.go

@@ -220,7 +220,7 @@ func SetDispatcherTaskScheduler(taskScheduler threadpool.ThreadPool) {
 	DispatcherTaskScheduler = taskScheduler
 }
 
-// EventsHandler is used to dispatcher the events received.
+// EventsHandler is used to dispatch the received events.
 // If the event is a DML event, it will be added to the sink for writing to downstream.
 // If the event is a resolved TS event, it will be update the resolvedTs of the dispatcher.
 // If the event is a DDL event,

logservice/eventstore/event_store.go

@@ -542,7 +542,7 @@ func (e *eventStore) updateMetrics(ctx context.Context) error {
 			e.dispatcherStates.RLock()
 			for _, subscriptionStat := range e.dispatcherStates.n {
 				// resolved ts lag
-				resolvedTs := subscriptionStat.resolvedTs
+				resolvedTs := atomic.LoadUint64(&subscriptionStat.resolvedTs)
 				resolvedPhyTs := oracle.ExtractPhysical(resolvedTs)
 				resolvedLag := float64(currentPhyTs-resolvedPhyTs) / 1e3
 				metrics.EventStoreDispatcherResolvedTsLagHist.Observe(float64(resolvedLag))
@@ -616,7 +616,7 @@ func (e *eventStore) batchAndWriteEvents(ctx context.Context, db *pebble.DB, inp
 					log.Warn("unknown subscriptionID", zap.Uint64("subID", uint64(subID)))
 					continue
 				}
-				subscriptionStat.resolvedTs = resolvedTs
+				atomic.StoreUint64(&subscriptionStat.resolvedTs, resolvedTs)
 				for dispatcherID := range subscriptionStat.ids {
 					dispatcherStat := e.dispatcherStates.m[dispatcherID]
 					dispatcherStat.notifier(resolvedTs)

maintainer/maintainer.go

@@ -614,13 +614,11 @@ func (m *Maintainer) getNewBootstrapFn() bootstrap.NewBootstrapMessageFn {
 			zap.Error(err))
 	}
 	return func(id node.ID) *messaging.TargetMessage {
-		var ddlDispatcherID *heartbeatpb.DispatcherID
 		// only send dispatcher id to dispatcher manager on the same node
 		if id == m.selfNode.ID {
-			ddlDispatcherID = m.tableTriggerEventDispatcherID.ToPB()
 			log.Info("create table event trigger dispatcher", zap.String("changefeed", m.id.String()),
 				zap.String("server", id.String()),
-				zap.String("dispatcher id", ddlDispatcherID.String()))
+				zap.String("dispatcher id", m.tableTriggerEventDispatcherID.String()))
 		}
 		log.Info("send maintainer bootstrap message",
 			zap.String("changefeed", m.id.String()),
@@ -633,7 +631,7 @@ func (m *Maintainer) getNewBootstrapFn() bootstrap.NewBootstrapMessageFn {
 				ChangefeedID:                  m.id.ToPB(),
 				Config:                        cfgBytes,
 				StartTs:                       m.startCheckpointTs,
-				TableTriggerEventDispatcherId: ddlDispatcherID,
+				TableTriggerEventDispatcherId: m.tableTriggerEventDispatcherID.ToPB(),
 			})
 	}
 }