tp: support per-table checkpoint (pingcap#5709)

Signed-off-by: Neil Shen <[email protected]>
overvenus · Jun 23, 2022 · d84f7fe · d84f7fe
1 parent 3f5e584
commit d84f7fe
Show file tree

Hide file tree

Showing 11 changed files with 530 additions and 253 deletions.
diff --git a/cdc/scheduler/internal/tp/agent.go b/cdc/scheduler/internal/tp/agent.go
@@ -311,7 +311,7 @@ func (a *agent) handleMessageDispatchTableRequest(
 		}
 		task = &dispatchTableTask{
 			TableID:   req.AddTable.GetTableID(),
-			StartTs:   req.AddTable.GetCheckpoint().GetCheckpointTs(),
+			StartTs:   req.AddTable.GetCheckpoint().CheckpointTs,
 			IsRemove:  false,
 			IsPrepare: req.AddTable.GetIsSecondary(),
 			Epoch:     epoch,
@@ -387,7 +387,7 @@ func (a *agent) newRemoveTableResponseMessage(
 			Response: &schedulepb.DispatchTableResponse_RemoveTable{
 				RemoveTable: &schedulepb.RemoveTableResponse{
 					Status:     &status,
-					Checkpoint: &status.Checkpoint,
+					Checkpoint: status.Checkpoint,
 				},
 			},
 		},
@@ -453,7 +453,7 @@ func (a *agent) newAddTableResponseMessage(
 			Response: &schedulepb.DispatchTableResponse_AddTable{
 				AddTable: &schedulepb.AddTableResponse{
 					Status:     &status,
-					Checkpoint: &status.Checkpoint,
+					Checkpoint: status.Checkpoint,
 					Reject:     reject,
 				},
 			},

diff --git a/cdc/scheduler/internal/tp/agent_test.go b/cdc/scheduler/internal/tp/agent_test.go
@@ -77,7 +77,6 @@ func TestAgentCollectTableStatus(t *testing.T) {
 	a.runningTasks[model.TableID(0)] = &dispatchTableTask{IsRemove: true}
 	status := a.newTableStatus(model.TableID(0))
 	require.Equal(t, schedulepb.TableStateStopping, status.State)
-
 }
 
 func TestAgentHandleDispatchTableTask(t *testing.T) {
@@ -221,13 +220,13 @@ func TestAgentHandleMessageStopping(t *testing.T) {
 				AddTable: &schedulepb.AddTableRequest{
 					TableID:     1,
 					IsSecondary: true,
-					Checkpoint:  &schedulepb.Checkpoint{},
 				},
 			},
 		},
 	}
 	// add table request should not be handled, so the running task count is 0.
 	response = a.handleMessage([]*schedulepb.Message{addTableRequest})
+	require.Len(t, response, 0)
 	require.Len(t, a.runningTasks, 0)
 
 	// mock agent have running task before stopping but processed yet.
@@ -288,7 +287,7 @@ func TestAgentHandleMessage(t *testing.T) {
 				AddTable: &schedulepb.AddTableRequest{
 					TableID:     1,
 					IsSecondary: true,
-					Checkpoint:  &schedulepb.Checkpoint{},
+					Checkpoint:  schedulepb.Checkpoint{},
 				},
 			},
 		},
@@ -392,7 +391,7 @@ func TestAgentTick(t *testing.T) {
 				AddTable: &schedulepb.AddTableRequest{
 					TableID:     1,
 					IsSecondary: true,
-					Checkpoint:  &schedulepb.Checkpoint{},
+					Checkpoint:  schedulepb.Checkpoint{},
 				},
 			},
 		},

diff --git a/cdc/scheduler/internal/tp/capture_manager.go b/cdc/scheduler/internal/tp/capture_manager.go
@@ -100,10 +100,6 @@ func newCaptureManager(rev schedulepb.OwnerRevision, heartbeatTick int) *capture
 	}
 }
 
-func (c *captureManager) CaptureTableSets() map[model.CaptureID]*CaptureStatus {
-	return c.Captures
-}
-
 func (c *captureManager) CheckAllCaptureInitialized() bool {
 	if !c.checkAllCaptureInitialized() {
 		return false

diff --git a/cdc/scheduler/internal/tp/coordinator.go b/cdc/scheduler/internal/tp/coordinator.go
@@ -27,6 +27,8 @@ import (
 	"go.uber.org/zap"
 )
 
+const checkpointCannotProceed = internal.CheckpointCannotProceed
+
 type scheduler interface {
 	Name() string
 	Schedule(
@@ -82,11 +84,7 @@ func (c *coordinator) Tick(
 	// All captures that are alive according to the latest Etcd states.
 	aliveCaptures map[model.CaptureID]*model.CaptureInfo,
 ) (newCheckpointTs, newResolvedTs model.Ts, err error) {
-	err = c.poll(ctx, checkpointTs, currentTables, aliveCaptures)
-	if err != nil {
-		return internal.CheckpointCannotProceed, internal.CheckpointCannotProceed, errors.Trace(err)
-	}
-	return internal.CheckpointCannotProceed, internal.CheckpointCannotProceed, nil
+	return c.poll(ctx, checkpointTs, currentTables, aliveCaptures)
 }
 
 func (c *coordinator) MoveTable(tableID model.TableID, target model.CaptureID) {}
@@ -100,10 +98,10 @@ func (c *coordinator) Close(ctx context.Context) {}
 func (c *coordinator) poll(
 	ctx context.Context, checkpointTs model.Ts, currentTables []model.TableID,
 	aliveCaptures map[model.CaptureID]*model.CaptureInfo,
-) error {
+) (newCheckpointTs, newResolvedTs model.Ts, err error) {
 	recvMsgs, err := c.recvMsgs(ctx)
 	if err != nil {
-		return errors.Trace(err)
+		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
 	}
 
 	var msgBuf []*schedulepb.Message
@@ -112,25 +110,25 @@ func (c *coordinator) poll(
 	msgBuf = append(msgBuf, msgs...)
 	msgs = c.captureM.HandleAliveCaptureUpdate(aliveCaptures)
 	msgBuf = append(msgBuf, msgs...)
-	if c.captureM.CheckAllCaptureInitialized() {
+	if !c.captureM.CheckAllCaptureInitialized() {
 		// Skip handling messages and tasks for replication manager,
 		// as not all capture are initialized.
-		return c.sendMsgs(ctx, msgBuf)
+		return checkpointCannotProceed, checkpointCannotProceed, c.sendMsgs(ctx, msgBuf)
 	}
 
 	// Handle capture membership changes.
 	if changes := c.captureM.TakeChanges(); changes != nil {
-		msgs, err = c.replicationM.HandleCaptureChanges(changes)
+		msgs, err = c.replicationM.HandleCaptureChanges(changes, checkpointTs)
 		if err != nil {
-			return errors.Trace(err)
+			return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
 		}
 		msgBuf = append(msgBuf, msgs...)
 	}
 
 	// Handle received messages to advance replication set.
 	msgs, err = c.replicationM.HandleMessage(recvMsgs)
 	if err != nil {
-		return errors.Trace(err)
+		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
 	}
 	msgBuf = append(msgBuf, msgs...)
 
@@ -150,18 +148,19 @@ func (c *coordinator) poll(
 	// Handle generated schedule tasks.
 	msgs, err = c.replicationM.HandleTasks(allTasks)
 	if err != nil {
-		return errors.Trace(err)
+		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
 	}
 	msgBuf = append(msgBuf, msgs...)
 
 	// Send new messages.
 	err = c.sendMsgs(ctx, msgBuf)
 	if err != nil {
-		return errors.Trace(err)
+		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
 	}
 
-	// checkpoint calculation
-	return nil
+	// Checkpoint calculation
+	newCheckpointTs, newResolvedTs = c.replicationM.AdvanceCheckpoint(currentTables)
+	return newCheckpointTs, newResolvedTs, nil
 }
 
 func (c *coordinator) recvMsgs(ctx context.Context) ([]*schedulepb.Message, error) {