tp: cleanup changes to schedulerv2 (#5975)

Signed-off-by: Neil Shen <[email protected]>
pingcap · Jun 23, 2022 · 60898f6 · 60898f6
1 parent 19d2492
commit 60898f6
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 29 deletions.
diff --git a/cdc/scheduler/internal/base/agent.go b/cdc/scheduler/internal/base/agent.go
@@ -37,8 +37,7 @@ import (
 type ProcessorMessenger interface {
 	// FinishTableOperation notifies the owner that a table operation has finished.
 	FinishTableOperation(
-		ctx context.Context, tableID model.TableID, checkpointTs model.Ts,
-		epoch protocol.ProcessorEpoch,
+		ctx context.Context, tableID model.TableID, epoch protocol.ProcessorEpoch,
 	) (done bool, err error)
 	// SyncTaskStatuses informs the owner of the processor's current internal state.
 	SyncTaskStatuses(
@@ -143,11 +142,10 @@ const (
 )
 
 type agentOperation struct {
-	TableID   model.TableID
-	StartTs   model.Ts
-	IsDelete  bool
-	IsPrepare bool
-	Epoch     protocol.ProcessorEpoch
+	TableID  model.TableID
+	StartTs  model.Ts
+	IsDelete bool
+	Epoch    protocol.ProcessorEpoch
 
 	// FromOwnerID is for debugging purposesFromOwnerID
 	FromOwnerID model.CaptureID
@@ -266,18 +264,14 @@ func (a *Agent) sendSync(ctx context.Context) (bool, error) {
 
 // processOperations tries to make progress on each pending table operations.
 // It queries the executor for the current status of each table.
-func (a *Agent) processOperations(ctx context.Context) (err error) {
+func (a *Agent) processOperations(ctx context.Context) error {
 	for tableID, op := range a.tableOperations {
-		var (
-			done         bool
-			checkpointTs model.Ts
-		)
 		switch op.status {
 		case operationReceived:
 			a.logger.Info("Agent start processing operation", zap.Any("op", op))
 			if !op.IsDelete {
 				// add table
-				done, err = a.executor.AddTable(ctx, op.TableID, op.StartTs, op.IsPrepare)
+				done, err := a.executor.AddTable(ctx, op.TableID, op.StartTs, false)
 				if err != nil {
 					return errors.Trace(err)
 				}
@@ -286,18 +280,19 @@ func (a *Agent) processOperations(ctx context.Context) (err error) {
 				}
 			} else {
 				// delete table
-				done = a.executor.RemoveTable(ctx, op.TableID)
+				done := a.executor.RemoveTable(ctx, op.TableID)
 				if !done {
 					break
 				}
 			}
 			op.status = operationProcessed
 			fallthrough
 		case operationProcessed:
+			var done bool
 			if !op.IsDelete {
-				done = a.executor.IsAddTableFinished(ctx, op.TableID, op.IsPrepare)
+				done = a.executor.IsAddTableFinished(ctx, op.TableID, false)
 			} else {
-				checkpointTs, done = a.executor.IsRemoveTableFinished(ctx, op.TableID)
+				_, done = a.executor.IsRemoveTableFinished(ctx, op.TableID)
 			}
 			if !done {
 				break
@@ -306,7 +301,7 @@ func (a *Agent) processOperations(ctx context.Context) (err error) {
 			fallthrough
 		case operationFinished:
 			a.logger.Info("Agent finish processing operation", zap.Any("op", op))
-			done, err = a.communicator.FinishTableOperation(ctx, op.TableID, checkpointTs, a.getEpoch())
+			done, err := a.communicator.FinishTableOperation(ctx, op.TableID, a.getEpoch())
 			if err != nil {
 				return errors.Trace(err)
 			}

diff --git a/cdc/scheduler/internal/base/agent_mock.go b/cdc/scheduler/internal/base/agent_mock.go
@@ -33,7 +33,7 @@ type MockProcessorMessenger struct {
 
 // FinishTableOperation marks this function as being called.
 func (m *MockProcessorMessenger) FinishTableOperation(
-	ctx context.Context, tableID model.TableID, checkpointTs model.Ts, epoch protocol.ProcessorEpoch,
+	ctx context.Context, tableID model.TableID, epoch protocol.ProcessorEpoch,
 ) (bool, error) {
 	args := m.Called(ctx, tableID, epoch)
 	return args.Bool(0), args.Error(1)
@@ -153,9 +153,6 @@ func (e *MockTableExecutor) IsAddTableFinished(ctx context.Context, tableID mode
 }
 
 // IsRemoveTableFinished determines if the table has been removed.
-// func (e *MockTableExecutor) IsRemoveTableFinished(
-// 	ctx context.Context, tableID model.TableID,
-// ) (model.Ts, bool) {
 func (e *MockTableExecutor) IsRemoveTableFinished(ctx context.Context, tableID model.TableID) (model.Ts, bool) {
 	_, ok := e.Removing[tableID]
 	return 0, !ok

diff --git a/cdc/scheduler/internal/base/processor_agent.go b/cdc/scheduler/internal/base/processor_agent.go
@@ -180,7 +180,7 @@ func (a *agentImpl) Tick(ctx context.Context) error {
 
 func (a *agentImpl) FinishTableOperation(
 	ctx context.Context,
-	tableID model.TableID, checkpointTs model.Ts,
+	tableID model.TableID,
 	epoch protocol.ProcessorEpoch,
 ) (done bool, err error) {
 	topic := protocol.SyncTopic(a.changeFeed)
@@ -196,11 +196,7 @@ func (a *agentImpl) FinishTableOperation(
 		}
 	}
 
-	message := &protocol.DispatchTableResponseMessage{
-		ID:           tableID,
-		Epoch:        epoch,
-		CheckpointTs: checkpointTs,
-	}
+	message := &protocol.DispatchTableResponseMessage{ID: tableID, Epoch: epoch}
 	defer func() {
 		if err != nil {
 			return

diff --git a/cdc/scheduler/internal/base/protocol/protocol.go b/cdc/scheduler/internal/base/protocol/protocol.go
@@ -49,8 +49,6 @@ func DispatchTableResponseTopic(changefeedID model.ChangeFeedID) p2p.Topic {
 type DispatchTableResponseMessage struct {
 	ID    model.TableID  `json:"id"`
 	Epoch ProcessorEpoch `json:"epoch"`
-
-	CheckpointTs model.Ts `json:"checkpoint"`
 }
 
 // AnnounceTopic returns a message topic for announcing an ownership change.