cdc: implement replication set (pingcap#5450)

Signed-off-by: Neil Shen <overvenus@gmail.com>
overvenus · Jun 23, 2022 · 677031c · 677031c
1 parent 09cf059
commit 677031c
Show file tree

Hide file tree

Showing 7 changed files with 1,289 additions and 151 deletions.
diff --git a/cdc/scheduler/internal/tp/balance_scheduler.go b/cdc/scheduler/internal/tp/balance_scheduler.go
@@ -28,6 +28,7 @@ func (b *balancer) Name() string {
 }
 
 func (b *balancer) Schedule(
+	checkpointTs model.Ts,
 	currentTables []model.TableID,
 	captures map[model.CaptureID]*model.CaptureInfo,
 	captureTables map[model.CaptureID]captureStatus,

diff --git a/cdc/scheduler/internal/tp/coordinator.go b/cdc/scheduler/internal/tp/coordinator.go
@@ -25,6 +25,7 @@ import (
 type scheduler interface {
 	Name() string
 	Schedule(
+		checkpointTs model.Ts,
 		currentTables []model.TableID,
 		aliveCaptures map[model.CaptureID]*model.CaptureInfo,
 		captureTables map[model.CaptureID]captureStatus,
@@ -71,7 +72,7 @@ func (c *coordinator) poll(
 	captureTables := c.manager.captureTableSets()
 	allTasks := make([]*scheduleTask, 0)
 	for _, sched := range c.scheduler {
-		tasks := sched.Schedule(currentTables, aliveCaptures, captureTables)
+		tasks := sched.Schedule(checkpointTs, currentTables, aliveCaptures, captureTables)
 		allTasks = append(allTasks, tasks...)
 	}
 	recvMsgs := c.recvMessages()
@@ -81,6 +82,8 @@ func (c *coordinator) poll(
 		return errors.Trace(err)
 	}
 	c.sendMessages(sentMsgs)
+
+	// checkpoint calcuation
 	return nil
 }