[DNM]: batch dispatch table message

cdc/scheduler/internal/v3/agent/agent.go

@@ -342,53 +342,51 @@ func (a *agent) handleMessageDispatchTableRequest(
 			zap.String("expected", a.Epoch.Epoch))
 		return
 	}
+
 	var (
 		table *tableSpan
 		task  *dispatchTableTask
 		ok    bool
 	)
-	// make the assumption that all tables are tracked by the agent now.
-	// this should be guaranteed by the caller of the method.
-	switch req := request.Request.(type) {
-	case *schedulepb.DispatchTableRequest_AddTable:
-		span := req.AddTable.GetSpan()
-		task = &dispatchTableTask{
-			Span:       span,
-			Checkpoint: req.AddTable.GetCheckpoint(),
-			IsRemove:   false,
-			IsPrepare:  req.AddTable.GetIsSecondary(),
-			Epoch:      epoch,
-			status:     dispatchTableTaskReceived,
-		}
-		table = a.tableM.addTableSpan(span)
-	case *schedulepb.DispatchTableRequest_RemoveTable:
-		span := req.RemoveTable.GetSpan()
-		table, ok = a.tableM.getTableSpan(span)
-		if !ok {
-			log.Warn("schedulerv3: agent ignore remove table request, "+
-				"since the table not found",
-				zap.String("capture", a.CaptureID),
-				zap.String("namespace", a.ChangeFeedID.Namespace),
-				zap.String("changefeed", a.ChangeFeedID.ID),
-				zap.String("span", span.String()),
-				zap.Any("request", request))
-			return
+
+	if request.GetBatchRemove() != nil {
+		for _, req := range request.GetBatchRemove().Requests {
+			span := req.GetSpan()
+			table, ok = a.tableM.getTableSpan(span)
+			if !ok {
+				log.Warn("schedulerv3: agent ignore remove table request, "+
+					"since the table not found",
+					zap.String("capture", a.CaptureID),
+					zap.String("namespace", a.ChangeFeedID.Namespace),
+					zap.String("changefeed", a.ChangeFeedID.ID),
+					zap.String("span", span.String()),
+					zap.Any("request", request))
+				return
+			}
+			task = &dispatchTableTask{
+				Span:     span,
+				IsRemove: true,
+				Epoch:    epoch,
+				status:   dispatchTableTaskReceived,
+			}
+			table.injectDispatchTableTask(task)
 		}
-		task = &dispatchTableTask{
-			Span:     span,
-			IsRemove: true,
-			Epoch:    epoch,
-			status:   dispatchTableTaskReceived,
+	}
+	if request.GetBatchAdd() != nil {
+		for _, req := range request.GetBatchAdd().Requests {
+			span := req.GetSpan()
+			task = &dispatchTableTask{
+				Span:       span,
+				Checkpoint: req.GetCheckpoint(),
+				IsRemove:   false,
+				IsPrepare:  req.GetIsSecondary(),
+				Epoch:      epoch,
+				status:     dispatchTableTaskReceived,
+			}
+			table = a.tableM.addTableSpan(span)
+			table.injectDispatchTableTask(task)
 		}
-	default:
-		log.Warn("schedulerv3: agent ignore unknown dispatch table request",
-			zap.String("capture", a.CaptureID),
-			zap.String("namespace", a.ChangeFeedID.Namespace),
-			zap.String("changefeed", a.ChangeFeedID.ID),
-			zap.Any("request", request))
-		return
 	}
-	table.injectDispatchTableTask(task)
 }
 
 // Close implement agent interface

cdc/scheduler/internal/v3/agent/table.go

@@ -327,7 +327,7 @@ func (tm *tableSpanManager) poll(ctx context.Context) ([]*schedulepb.Message, er
 		if message == nil {
 			return true
 		}
-		result = append(result, message)
+		// result = append(result, message)
 		return true
 	})
 	for _, span := range toBeDropped {

cdc/scheduler/internal/v3/coordinator.go

@@ -439,7 +439,49 @@ func (c *coordinator) sendMsgs(ctx context.Context, msgs []*schedulepb.Message)
 		m.From = c.captureID
 	}
 	c.compat.BeforeTransportSend(msgs)
-	return c.trans.Send(ctx, msgs)
+	var compatedMsgs []*schedulepb.Message
+	addTableMsgs := make(map[model.CaptureID]*schedulepb.Message)
+	removeTableMsgs := make(map[model.CaptureID]*schedulepb.Message)
+	for _, msg := range msgs {
+		switch msg.MsgType {
+		case schedulepb.MsgDispatchTableRequest:
+			switch req := msg.DispatchTableRequest.Request.(type) {
+			case *schedulepb.DispatchTableRequest_AddTable:
+				addTableReq := req.AddTable
+				oldMsg, ok := addTableMsgs[msg.To]
+				if !ok {
+					oldMsg = msg
+					addTableMsgs[msg.To] = msg
+					msg.DispatchTableRequest.Request = &schedulepb.DispatchTableRequest_BatchAdd{
+						BatchAdd: &schedulepb.AddTableRequests{
+							Requests: make([]*schedulepb.AddTableRequest, 0),
+						},
+					}
+					compatedMsgs = append(compatedMsgs, msg)
+				}
+				req.AddTable = nil
+				oldMsg.GetDispatchTableRequest().GetBatchAdd().Requests = append(oldMsg.GetDispatchTableRequest().GetBatchAdd().Requests, addTableReq)
+			case *schedulepb.DispatchTableRequest_RemoveTable:
+				removeTableReq := req.RemoveTable
+				oldMsg, ok := removeTableMsgs[msg.To]
+				if !ok {
+					oldMsg = msg
+					removeTableMsgs[msg.To] = msg
+					msg.DispatchTableRequest.Request = &schedulepb.DispatchTableRequest_BatchRemove{
+						BatchRemove: &schedulepb.RemoveTableRequests{
+							Requests: make([]*schedulepb.RemoveTableRequest, 0),
+						},
+					}
+					compatedMsgs = append(compatedMsgs, msg)
+				}
+				req.RemoveTable = nil
+				oldMsg.GetDispatchTableRequest().GetBatchRemove().Requests = append(oldMsg.GetDispatchTableRequest().GetBatchRemove().Requests, removeTableReq)
+			}
+		default:
+			compatedMsgs = append(compatedMsgs, msg)
+		}
+	}
+	return c.trans.Send(ctx, compatedMsgs)
 }
 
 func (c *coordinator) maybeCollectMetrics() {