update

Author: hongyunyan

downstreamadapter/dispatcher/helper.go

@@ -223,9 +223,11 @@ func NewDispatcherEvent(event commonEvent.Event) *DispatcherEvent {
 	}
 	switch event.GetType() {
 	case commonEvent.TypeResolvedEvent:
+		dispatcherEvent.isBatchable = true
 	case commonEvent.TypeDMLEvent:
 		dispatcherEvent.isBatchable = true
 	case commonEvent.TypeDDLEvent:
+		dispatcherEvent.isBatchable = false
 	case commonEvent.TypeSyncPointEvent:
 		dispatcherEvent.isBatchable = false
 	default:
@@ -240,7 +242,7 @@ var dispatcherEventsDynamicStreamOnce sync.Once
 func GetDispatcherEventsDynamicStream() dynstream.DynamicStream[common.DispatcherID, DispatcherEvent, *Dispatcher] {
 	if dispatcherEventsDynamicStream == nil {
 		dispatcherEventsDynamicStreamOnce.Do(func() {
-			dispatcherEventsDynamicStream = dynstream.NewDynamicStream(&DispatcherEventsHandler{})
+			dispatcherEventsDynamicStream = dynstream.NewDynamicStream(&DispatcherEventsHandler{}, dynstream.NewOptionWithBatchSize(128))
 			dispatcherEventsDynamicStream.Start()
 		})
 	}

utils/dynstream/interfaces.go

@@ -116,6 +116,12 @@ func NewOption() Option {
 	}
 }
 
+func NewOptionWithBatchSize(batchSize int) Option {
+	opt := NewOption()
+	opt.BatchSize = batchSize
+	return opt
+}
+
 func (o *Option) fix() {
 	if o.StreamCount == 0 {
 		o.StreamCount = runtime.NumCPU()

utils/dynstream/stream.go

@@ -344,19 +344,19 @@ Loop:
 						// which is possible when the path is removed or recovered from blocked.
 						break
 					}
-					// If the event is non-batchable, we should handle it singly.
-					// The non-batchable event should be the only event in a batch.
-					if !e.IsBatchable() {
+
+					if e.IsBatchable() {
+						signal.pathInfo.pendingQueue.PopFront()
+						eventBuf = append(eventBuf, e)
+					} else {
+						// If the event is non-batchable, we should handle it singly.
+						// The non-batchable event should be the only event in a batch.
 						if i == 0 {
 							eventBuf = append(eventBuf, e)
 							signal.pathInfo.pendingQueue.PopFront()
-							break
-						} else {
-							break
 						}
+						break
 					}
-					signal.pathInfo.pendingQueue.PopFront()
-					eventBuf = append(eventBuf, e)
 				}
 
 				actualCount := len(eventBuf)