From 77134524184c8ae11f168733786176b32b4ca128 Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Wed, 20 Apr 2022 20:05:23 +0800 Subject: [PATCH 01/13] try fix scheduler problem Signed-off-by: zeminzhou --- cdc/cdc/model/owner.go | 11 ++++--- cdc/cdc/owner/scheduler_v1.go | 54 +++++++++++++++++++++++++++------- cdc/cdc/processor/processor.go | 21 +++++++++++++ 3 files changed, 71 insertions(+), 15 deletions(-) diff --git a/cdc/cdc/model/owner.go b/cdc/cdc/model/owner.go index 791d3b18..19a4b83b 100644 --- a/cdc/cdc/model/owner.go +++ b/cdc/cdc/model/owner.go @@ -165,6 +165,8 @@ type KeySpanOperation struct { // if the operation is a add operation, BoundaryTs is start ts BoundaryTs uint64 `json:"boundary_ts"` Status uint64 `json:"status,omitempty"` + + RelatedKeySpans []uint64 `json:"related_key_spans"` } // KeySpanProcessed returns whether the keyspan has been processed by processor @@ -270,7 +272,7 @@ func (ts *TaskStatus) RemoveKeySpan(id KeySpanID, boundaryTs Ts, isMoveKeySpan b } // AddKeySpan add the keyspan in KeySpanInfos and add a add kyespan operation. -func (ts *TaskStatus) AddKeySpan(id KeySpanID, keyspan *KeySpanReplicaInfo, boundaryTs Ts) { +func (ts *TaskStatus) AddKeySpan(id KeySpanID, keyspan *KeySpanReplicaInfo, boundaryTs Ts, relatedKeySpans []KeySpanID) { if ts.KeySpans == nil { ts.KeySpans = make(map[KeySpanID]*KeySpanReplicaInfo) } @@ -284,9 +286,10 @@ func (ts *TaskStatus) AddKeySpan(id KeySpanID, keyspan *KeySpanReplicaInfo, boun ts.Operation = make(map[KeySpanID]*KeySpanOperation) } ts.Operation[id] = &KeySpanOperation{ - Delete: false, - BoundaryTs: boundaryTs, - Status: OperDispatched, + Delete: false, + BoundaryTs: boundaryTs, + Status: OperDispatched, + RelatedKeySpans: relatedKeySpans, } } diff --git a/cdc/cdc/owner/scheduler_v1.go b/cdc/cdc/owner/scheduler_v1.go index 90e9ca20..f29fde05 100644 --- a/cdc/cdc/owner/scheduler_v1.go +++ b/cdc/cdc/owner/scheduler_v1.go @@ -45,6 +45,8 @@ type schedulerJob struct { // if the operation is an add operation, boundaryTs is start ts BoundaryTs uint64 TargetCapture model.CaptureID + + RelatedKeySpans []model.KeySpanID } type moveKeySpanJob struct { @@ -54,7 +56,7 @@ type moveKeySpanJob struct { type oldScheduler struct { state *orchestrator.ChangefeedReactorState - currentKeySpansID []model.KeySpanID + currentKeySpanIDs []model.KeySpanID currentKeySpans map[model.KeySpanID]regionspan.Span captures map[model.CaptureID]*model.CaptureInfo @@ -84,13 +86,16 @@ func (s *oldScheduler) Tick( ) (shouldUpdateState bool, err error) { s.state = state - s.currentKeySpansID, s.currentKeySpans, err = s.updateCurrentKeySpans(ctx) + currentKeySpanIDs, currentKeySpans, err := s.updateCurrentKeySpans(ctx) if err != nil { return false, errors.Trace(err) } + relatedKeySpans := s.computeRelatedKeySpans(currentKeySpans) + s.currentKeySpanIDs, s.currentKeySpans = currentKeySpanIDs, currentKeySpans + s.cleanUpFinishedOperations() - pendingJob, err := s.syncKeySpansWithCurrentKeySpans() + pendingJob, err := s.syncKeySpansWithCurrentKeySpans(relatedKeySpans) if err != nil { return false, errors.Trace(err) } @@ -113,6 +118,32 @@ func (s *oldScheduler) Tick( return shouldUpdateState, nil } +func (s *oldScheduler) computeRelatedKeySpans(currentKeySpans map[model.KeySpanID]regionspan.Span) map[model.KeySpanID][]model.KeySpanID { + oldKeySpans := s.currentKeySpans + + newKeySpans := []model.KeySpanID{} + needRemovedKeySpans := []model.KeySpanID{} + + for keyspanID := range oldKeySpans { + if _, ok := currentKeySpans[keyspanID]; !ok { + needRemovedKeySpans = append(needRemovedKeySpans, keyspanID) + } + } + + for keyspanID := range currentKeySpans { + if _, ok := oldKeySpans[keyspanID]; !ok { + newKeySpans = append(newKeySpans, keyspanID) + } + } + + relatedKeySpans := map[model.KeySpanID][]model.KeySpanID{} + for _, keyspanID := range newKeySpans { + relatedKeySpans[keyspanID] = needRemovedKeySpans + } + + return relatedKeySpans +} + func (s *oldScheduler) MoveKeySpan(keyspanID model.KeySpanID, target model.CaptureID) { s.moveKeySpanJobQueue = append(s.moveKeySpanJobQueue, &moveKeySpanJob{ keyspanID: keyspanID, @@ -246,25 +277,26 @@ func (s *oldScheduler) dispatchToTargetCaptures(pendingJobs []*schedulerJob) { // syncKeySpansWithCurrentKeySpans iterates all current keyspans to check whether it should be listened or not. // this function will return schedulerJob to make sure all keyspans will be listened. -func (s *oldScheduler) syncKeySpansWithCurrentKeySpans() ([]*schedulerJob, error) { +func (s *oldScheduler) syncKeySpansWithCurrentKeySpans(relatedKeySpans map[model.KeySpanID][]model.KeySpanID) ([]*schedulerJob, error) { var pendingJob []*schedulerJob allKeySpanListeningNow, err := s.keyspan2CaptureIndex() if err != nil { return nil, errors.Trace(err) } globalCheckpointTs := s.state.Status.CheckpointTs - for _, keyspanID := range s.currentKeySpansID { + for _, keyspanID := range s.currentKeySpanIDs { if _, exist := allKeySpanListeningNow[keyspanID]; exist { delete(allKeySpanListeningNow, keyspanID) continue } // For each keyspan which should be listened but is not, add an adding-keyspan job to the pending job list pendingJob = append(pendingJob, &schedulerJob{ - Tp: schedulerJobTypeAddKeySpan, - KeySpanID: keyspanID, - Start: s.currentKeySpans[keyspanID].Start, - End: s.currentKeySpans[keyspanID].End, - BoundaryTs: globalCheckpointTs, + Tp: schedulerJobTypeAddKeySpan, + KeySpanID: keyspanID, + Start: s.currentKeySpans[keyspanID].Start, + End: s.currentKeySpans[keyspanID].End, + BoundaryTs: globalCheckpointTs, + RelatedKeySpans: relatedKeySpans[keyspanID], }) } // The remaining keyspans are the keyspans which should be not listened @@ -300,7 +332,7 @@ func (s *oldScheduler) handleJobs(jobs []*schedulerJob) { StartTs: job.BoundaryTs, Start: job.Start, End: job.End, - }, job.BoundaryTs) + }, job.BoundaryTs, job.RelatedKeySpans) case schedulerJobTypeRemoveKeySpan: failpoint.Inject("OwnerRemoveKeySpanError", func() { // just skip removing this keyspan diff --git a/cdc/cdc/processor/processor.go b/cdc/cdc/processor/processor.go index 32764fb6..56c8bcfe 100644 --- a/cdc/cdc/processor/processor.go +++ b/cdc/cdc/processor/processor.go @@ -574,6 +574,10 @@ func (p *processor) handleKeySpanOperation(ctx cdcContext.Context) error { if replicaInfo.StartTs != opt.BoundaryTs { log.Warn("the startTs and BoundaryTs of add keyspan operation should be always equaled", zap.Any("replicaInfo", replicaInfo)) } + + if !p.checkRelatedKeyspans(opt.RelatedKeySpans) { + continue + } err := p.addKeySpan(ctx, keyspanID, replicaInfo) if err != nil { return errors.Trace(err) @@ -613,6 +617,23 @@ func (p *processor) handleKeySpanOperation(ctx cdcContext.Context) error { return nil } +func (p *processor) checkRelatedKeyspans(relatedKeySpans []model.KeySpanID) bool { + + allTaskStatus := map[model.KeySpanID]uint64{} + for _, taskStatus := range p.changefeed.TaskStatuses { + for keyspanID, operation := range taskStatus.Operation { + allTaskStatus[keyspanID] = operation.Status + } + } + + for _, keyspanID := range relatedKeySpans { + if status, ok := allTaskStatus[keyspanID]; ok && status != model.OperFinished { + return false + } + } + return true +} + func (p *processor) sendError(err error) { if err == nil { return From 5a9e84ca05e508dd3c432ba8bc7e79e8e5f3f8ed Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Thu, 21 Apr 2022 11:57:09 +0800 Subject: [PATCH 02/13] optimization Signed-off-by: zeminzhou --- cdc/cdc/model/owner.go | 10 +++++-- cdc/cdc/owner/scheduler_v1.go | 55 ++++++++++++++++++++-------------- cdc/cdc/processor/processor.go | 17 ++++------- 3 files changed, 46 insertions(+), 36 deletions(-) diff --git a/cdc/cdc/model/owner.go b/cdc/cdc/model/owner.go index 19a4b83b..55df60ed 100644 --- a/cdc/cdc/model/owner.go +++ b/cdc/cdc/model/owner.go @@ -166,7 +166,7 @@ type KeySpanOperation struct { BoundaryTs uint64 `json:"boundary_ts"` Status uint64 `json:"status,omitempty"` - RelatedKeySpans []uint64 `json:"related_key_spans"` + RelatedKeySpans []KeySpanLocation `json:"related_key_spans"` } // KeySpanProcessed returns whether the keyspan has been processed by processor @@ -272,7 +272,7 @@ func (ts *TaskStatus) RemoveKeySpan(id KeySpanID, boundaryTs Ts, isMoveKeySpan b } // AddKeySpan add the keyspan in KeySpanInfos and add a add kyespan operation. -func (ts *TaskStatus) AddKeySpan(id KeySpanID, keyspan *KeySpanReplicaInfo, boundaryTs Ts, relatedKeySpans []KeySpanID) { +func (ts *TaskStatus) AddKeySpan(id KeySpanID, keyspan *KeySpanReplicaInfo, boundaryTs Ts, relatedKeySpans []KeySpanLocation) { if ts.KeySpans == nil { ts.KeySpans = make(map[KeySpanID]*KeySpanReplicaInfo) } @@ -451,3 +451,9 @@ type ProcInfoSnap struct { CaptureID string `json:"capture-id"` KeySpans map[KeySpanID]*KeySpanReplicaInfo `json:"-"` } + +// KeySpanLocation records which capture a keyspan is in +type KeySpanLocation struct { + CaptureID string `json:"capture_id"` + KeySpanID KeySpanID `json:"keyspan_id"` +} diff --git a/cdc/cdc/owner/scheduler_v1.go b/cdc/cdc/owner/scheduler_v1.go index f29fde05..a5ebfcbd 100644 --- a/cdc/cdc/owner/scheduler_v1.go +++ b/cdc/cdc/owner/scheduler_v1.go @@ -46,7 +46,7 @@ type schedulerJob struct { BoundaryTs uint64 TargetCapture model.CaptureID - RelatedKeySpans []model.KeySpanID + RelatedKeySpans []model.KeySpanLocation } type moveKeySpanJob struct { @@ -91,11 +91,11 @@ func (s *oldScheduler) Tick( return false, errors.Trace(err) } - relatedKeySpans := s.computeRelatedKeySpans(currentKeySpans) + newKeySpans, needRemoveKeySpans := s.diffCurrentKeySpans(currentKeySpans) s.currentKeySpanIDs, s.currentKeySpans = currentKeySpanIDs, currentKeySpans s.cleanUpFinishedOperations() - pendingJob, err := s.syncKeySpansWithCurrentKeySpans(relatedKeySpans) + pendingJob, err := s.syncKeySpansWithCurrentKeySpans(newKeySpans, needRemoveKeySpans) if err != nil { return false, errors.Trace(err) } @@ -118,30 +118,25 @@ func (s *oldScheduler) Tick( return shouldUpdateState, nil } -func (s *oldScheduler) computeRelatedKeySpans(currentKeySpans map[model.KeySpanID]regionspan.Span) map[model.KeySpanID][]model.KeySpanID { +func (s *oldScheduler) diffCurrentKeySpans(currentKeySpans map[model.KeySpanID]regionspan.Span) (map[model.KeySpanID]struct{}, []model.KeySpanID) { oldKeySpans := s.currentKeySpans - newKeySpans := []model.KeySpanID{} - needRemovedKeySpans := []model.KeySpanID{} + newKeySpans := map[model.KeySpanID]struct{}{} + needRemoveKeySpans := []model.KeySpanID{} for keyspanID := range oldKeySpans { if _, ok := currentKeySpans[keyspanID]; !ok { - needRemovedKeySpans = append(needRemovedKeySpans, keyspanID) + needRemoveKeySpans = append(needRemoveKeySpans, keyspanID) } } for keyspanID := range currentKeySpans { if _, ok := oldKeySpans[keyspanID]; !ok { - newKeySpans = append(newKeySpans, keyspanID) + newKeySpans[keyspanID] = struct{}{} } } - relatedKeySpans := map[model.KeySpanID][]model.KeySpanID{} - for _, keyspanID := range newKeySpans { - relatedKeySpans[keyspanID] = needRemovedKeySpans - } - - return relatedKeySpans + return newKeySpans, needRemoveKeySpans } func (s *oldScheduler) MoveKeySpan(keyspanID model.KeySpanID, target model.CaptureID) { @@ -277,27 +272,41 @@ func (s *oldScheduler) dispatchToTargetCaptures(pendingJobs []*schedulerJob) { // syncKeySpansWithCurrentKeySpans iterates all current keyspans to check whether it should be listened or not. // this function will return schedulerJob to make sure all keyspans will be listened. -func (s *oldScheduler) syncKeySpansWithCurrentKeySpans(relatedKeySpans map[model.KeySpanID][]model.KeySpanID) ([]*schedulerJob, error) { +func (s *oldScheduler) syncKeySpansWithCurrentKeySpans(newKeySpans map[model.KeySpanID]struct{}, needRemoveKeySpans []model.KeySpanID) ([]*schedulerJob, error) { var pendingJob []*schedulerJob allKeySpanListeningNow, err := s.keyspan2CaptureIndex() if err != nil { return nil, errors.Trace(err) } + relatedKeySpans := make([]model.KeySpanLocation, 0, len(needRemoveKeySpans)) + for _, keyspanID := range needRemoveKeySpans { + if captureID, ok := allKeySpanListeningNow[keyspanID]; ok { + location := model.KeySpanLocation{ + CaptureID: captureID, + KeySpanID: keyspanID, + } + relatedKeySpans = append(relatedKeySpans, location) + } + } + globalCheckpointTs := s.state.Status.CheckpointTs for _, keyspanID := range s.currentKeySpanIDs { if _, exist := allKeySpanListeningNow[keyspanID]; exist { delete(allKeySpanListeningNow, keyspanID) continue } + job := &schedulerJob{ + Tp: schedulerJobTypeAddKeySpan, + KeySpanID: keyspanID, + Start: s.currentKeySpans[keyspanID].Start, + End: s.currentKeySpans[keyspanID].End, + BoundaryTs: globalCheckpointTs, + } + if _, ok := newKeySpans[keyspanID]; ok { + job.RelatedKeySpans = relatedKeySpans + } // For each keyspan which should be listened but is not, add an adding-keyspan job to the pending job list - pendingJob = append(pendingJob, &schedulerJob{ - Tp: schedulerJobTypeAddKeySpan, - KeySpanID: keyspanID, - Start: s.currentKeySpans[keyspanID].Start, - End: s.currentKeySpans[keyspanID].End, - BoundaryTs: globalCheckpointTs, - RelatedKeySpans: relatedKeySpans[keyspanID], - }) + pendingJob = append(pendingJob, job) } // The remaining keyspans are the keyspans which should be not listened keyspansThatShouldNotBeListened := allKeySpanListeningNow diff --git a/cdc/cdc/processor/processor.go b/cdc/cdc/processor/processor.go index 56c8bcfe..2035093d 100644 --- a/cdc/cdc/processor/processor.go +++ b/cdc/cdc/processor/processor.go @@ -617,19 +617,14 @@ func (p *processor) handleKeySpanOperation(ctx cdcContext.Context) error { return nil } -func (p *processor) checkRelatedKeyspans(relatedKeySpans []model.KeySpanID) bool { - - allTaskStatus := map[model.KeySpanID]uint64{} - for _, taskStatus := range p.changefeed.TaskStatuses { - for keyspanID, operation := range taskStatus.Operation { - allTaskStatus[keyspanID] = operation.Status +func (p *processor) checkRelatedKeyspans(relatedKeySpans []model.KeySpanLocation) bool { + for _, location := range relatedKeySpans { + if taskStatus, ok := p.changefeed.TaskStatuses[location.CaptureID]; ok { + if operation, ok := taskStatus.Operation[location.KeySpanID]; ok && operation.Status != model.OperFinished { + return false + } } - } - for _, keyspanID := range relatedKeySpans { - if status, ok := allTaskStatus[keyspanID]; ok && status != model.OperFinished { - return false - } } return true } From ea8d92aaecd48aeb873b2a5a5eff6a3a3625af57 Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Mon, 25 Apr 2022 15:45:55 +0800 Subject: [PATCH 03/13] fix ut Signed-off-by: zeminzhou --- cdc/cdc/processor/processor_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/cdc/cdc/processor/processor_test.go b/cdc/cdc/processor/processor_test.go index c225d79a..93efc12d 100644 --- a/cdc/cdc/processor/processor_test.go +++ b/cdc/cdc/processor/processor_test.go @@ -217,7 +217,7 @@ func (s *processorSuite) TestHandleKeySpanOperation4SingleKeySpan(c *check.C) { // add keyspan, in processing // in current implementation of owner, the startTs and BoundaryTs of add keyspan operation should be always equaled. p.changefeed.PatchTaskStatus(p.captureInfo.ID, func(status *model.TaskStatus) (*model.TaskStatus, bool, error) { - status.AddKeySpan(66, &model.KeySpanReplicaInfo{StartTs: 60}, 60) + status.AddKeySpan(66, &model.KeySpanReplicaInfo{StartTs: 60}, 60, []model.KeySpanLocation{}) return status, true, nil }) tester.MustApplyPatches() @@ -351,9 +351,9 @@ func (s *processorSuite) TestHandleKeySpanOperation4MultiKeySpan(c *check.C) { // add keyspan, in processing // in current implementation of owner, the startTs and BoundaryTs of add keyspan operation should be always equaled. p.changefeed.PatchTaskStatus(p.captureInfo.ID, func(status *model.TaskStatus) (*model.TaskStatus, bool, error) { - status.AddKeySpan(1, &model.KeySpanReplicaInfo{StartTs: 60}, 60) - status.AddKeySpan(2, &model.KeySpanReplicaInfo{StartTs: 50}, 50) - status.AddKeySpan(3, &model.KeySpanReplicaInfo{StartTs: 40}, 40) + status.AddKeySpan(1, &model.KeySpanReplicaInfo{StartTs: 60}, 60, []model.KeySpanLocation{}) + status.AddKeySpan(2, &model.KeySpanReplicaInfo{StartTs: 50}, 50, []model.KeySpanLocation{}) + status.AddKeySpan(3, &model.KeySpanReplicaInfo{StartTs: 40}, 40, []model.KeySpanLocation{}) status.KeySpans[4] = &model.KeySpanReplicaInfo{StartTs: 30} return status, true, nil }) From 8365eceb7530f4cb656ac495cad399fb9c5ac1b2 Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Mon, 25 Apr 2022 16:03:28 +0800 Subject: [PATCH 04/13] fix ut Signed-off-by: zeminzhou --- cdc/cdc/model/owner_test.go | 23 ++++------------------- cdc/cdc/processor/processor_test.go | 8 ++++---- 2 files changed, 8 insertions(+), 23 deletions(-) diff --git a/cdc/cdc/model/owner_test.go b/cdc/cdc/model/owner_test.go index 2656eb89..77d303ac 100644 --- a/cdc/cdc/model/owner_test.go +++ b/cdc/cdc/model/owner_test.go @@ -48,21 +48,6 @@ func TestAdminJobType(t *testing.T) { } } -func TestDDLStateString(t *testing.T) { - t.Parallel() - - names := map[ChangeFeedDDLState]string{ - ChangeFeedSyncDML: "SyncDML", - ChangeFeedWaitToExecDDL: "WaitToExecDDL", - ChangeFeedExecDDL: "ExecDDL", - ChangeFeedDDLExecuteFailed: "DDLExecuteFailed", - ChangeFeedDDLState(100): "Unknown", - } - for state, name := range names { - require.Equal(t, name, state.String()) - } -} - func TestTaskPositionMarshal(t *testing.T) { t.Parallel() @@ -265,11 +250,11 @@ func TestAddKeySpan(t *testing.T) { }, } status := &TaskStatus{} - status.AddKeySpan(1, &KeySpanReplicaInfo{StartTs: ts}, ts) + status.AddKeySpan(1, &KeySpanReplicaInfo{StartTs: ts}, ts, nil) require.Equal(t, expected, status) // add existing keyspan does nothing - status.AddKeySpan(1, &KeySpanReplicaInfo{StartTs: 1}, 1) + status.AddKeySpan(1, &KeySpanReplicaInfo{StartTs: 1}, 1, nil) require.Equal(t, expected, status) } @@ -279,8 +264,8 @@ func TestTaskStatusApplyState(t *testing.T) { ts1 := uint64(420875042036766723) ts2 := uint64(420876783269969921) status := &TaskStatus{} - status.AddKeySpan(1, &KeySpanReplicaInfo{StartTs: ts1}, ts1) - status.AddKeySpan(2, &KeySpanReplicaInfo{StartTs: ts2}, ts2) + status.AddKeySpan(1, &KeySpanReplicaInfo{StartTs: ts1}, ts1, nil) + status.AddKeySpan(2, &KeySpanReplicaInfo{StartTs: ts2}, ts2, nil) require.True(t, status.SomeOperationsUnapplied()) require.Equal(t, ts1, status.AppliedTs()) diff --git a/cdc/cdc/processor/processor_test.go b/cdc/cdc/processor/processor_test.go index 93efc12d..0b1a21f3 100644 --- a/cdc/cdc/processor/processor_test.go +++ b/cdc/cdc/processor/processor_test.go @@ -217,7 +217,7 @@ func (s *processorSuite) TestHandleKeySpanOperation4SingleKeySpan(c *check.C) { // add keyspan, in processing // in current implementation of owner, the startTs and BoundaryTs of add keyspan operation should be always equaled. p.changefeed.PatchTaskStatus(p.captureInfo.ID, func(status *model.TaskStatus) (*model.TaskStatus, bool, error) { - status.AddKeySpan(66, &model.KeySpanReplicaInfo{StartTs: 60}, 60, []model.KeySpanLocation{}) + status.AddKeySpan(66, &model.KeySpanReplicaInfo{StartTs: 60}, 60, nil) return status, true, nil }) tester.MustApplyPatches() @@ -351,9 +351,9 @@ func (s *processorSuite) TestHandleKeySpanOperation4MultiKeySpan(c *check.C) { // add keyspan, in processing // in current implementation of owner, the startTs and BoundaryTs of add keyspan operation should be always equaled. p.changefeed.PatchTaskStatus(p.captureInfo.ID, func(status *model.TaskStatus) (*model.TaskStatus, bool, error) { - status.AddKeySpan(1, &model.KeySpanReplicaInfo{StartTs: 60}, 60, []model.KeySpanLocation{}) - status.AddKeySpan(2, &model.KeySpanReplicaInfo{StartTs: 50}, 50, []model.KeySpanLocation{}) - status.AddKeySpan(3, &model.KeySpanReplicaInfo{StartTs: 40}, 40, []model.KeySpanLocation{}) + status.AddKeySpan(1, &model.KeySpanReplicaInfo{StartTs: 60}, 60, nil) + status.AddKeySpan(2, &model.KeySpanReplicaInfo{StartTs: 50}, 50, nil) + status.AddKeySpan(3, &model.KeySpanReplicaInfo{StartTs: 40}, 40, nil) status.KeySpans[4] = &model.KeySpanReplicaInfo{StartTs: 30} return status, true, nil }) From ed5ea96cd45f2489e551c5ed932654a25839d702 Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Mon, 25 Apr 2022 16:39:31 +0800 Subject: [PATCH 05/13] remove commented codes Signed-off-by: zeminzhou --- cdc/cdc/server.go | 6 ------ 1 file changed, 6 deletions(-) diff --git a/cdc/cdc/server.go b/cdc/cdc/server.go index 689081ca..fab7f081 100644 --- a/cdc/cdc/server.go +++ b/cdc/cdc/server.go @@ -261,12 +261,6 @@ func (s *Server) run(ctx context.Context) (err error) { return s.etcdHealthChecker(cctx) }) - /* - wg.Go(func() error { - return unified.RunWorkerPool(cctx) - }) - */ - wg.Go(func() error { return kv.RunWorkerPool(cctx) }) From 7446ada323388d15c34e067997a74eae403407c6 Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Tue, 10 May 2022 13:22:56 +0800 Subject: [PATCH 06/13] fix ut for owner Signed-off-by: zeminzhou --- cdc/cdc/model/owner.go | 3 +- cdc/cdc/owner/changefeed.go | 246 +---------------------------- cdc/cdc/owner/changefeed_test.go | 5 +- cdc/cdc/owner/scheduler.go | 19 ++- cdc/cdc/owner/scheduler_test.go | 39 ++--- cdc/cdc/owner/scheduler_v1.go | 27 +++- cdc/cdc/owner/scheduler_v1_test.go | 164 +++++++++++++++---- 7 files changed, 191 insertions(+), 312 deletions(-) diff --git a/cdc/cdc/model/owner.go b/cdc/cdc/model/owner.go index 55df60ed..958c8398 100644 --- a/cdc/cdc/model/owner.go +++ b/cdc/cdc/model/owner.go @@ -195,7 +195,7 @@ type TaskWorkload map[KeySpanID]WorkloadInfo // WorkloadInfo records the workload info of a keyspan type WorkloadInfo struct { - Workload uint64 `json:"workload"` + Workload int64 `json:"workload"` } // Unmarshal unmarshals into *TaskWorkload from json marshal byte slice @@ -219,7 +219,6 @@ type KeySpanReplicaInfo struct { StartTs Ts `json:"start-ts"` Start []byte End []byte - // MarkKeySpanID KeySpanID `json:"mark-keyspan-id"` } // Clone clones a KeySpanReplicaInfo diff --git a/cdc/cdc/owner/changefeed.go b/cdc/cdc/owner/changefeed.go index 7ed144e5..c53f14a7 100644 --- a/cdc/cdc/owner/changefeed.go +++ b/cdc/cdc/owner/changefeed.go @@ -40,27 +40,14 @@ type changefeed struct { id model.ChangeFeedID state *orchestrator.ChangefeedReactorState - scheduler scheduler - // barriers *barriers + scheduler scheduler feedStateManager *feedStateManager gcManager gc.Manager - // TODO: Can we delete redoManager for tikv cdc? - // redoManager redo.LogManager - - // schema *schemaWrap4Owner - // sink DDLSink - // ddlPuller DDLPuller initialized bool // isRemoved is true if the changefeed is removed isRemoved bool - // only used for asyncExecDDL function - // ddlEventCache is not nil when the changefeed is executing a DDL event asynchronously - // After the DDL event has been executed, ddlEventCache will be set to nil. - - // ddlEventCache *model.DDLEvent - errCh chan error // cancel the running goroutine start by `DDLPuller` cancel context.CancelFunc @@ -75,8 +62,6 @@ type changefeed struct { metricsChangefeedResolvedTsGauge prometheus.Gauge metricsChangefeedResolvedTsLagGauge prometheus.Gauge - // newDDLPuller func(ctx cdcContext.Context, startTs uint64) (DDLPuller, error) - // newSink func() DDLSink newScheduler func(ctx cdcContext.Context, startTs uint64) (scheduler, error) } @@ -84,34 +69,22 @@ func newChangefeed(id model.ChangeFeedID, gcManager gc.Manager) *changefeed { c := &changefeed{ id: id, // The scheduler will be created lazily. - scheduler: nil, - // barriers: newBarriers(), + scheduler: nil, feedStateManager: new(feedStateManager), gcManager: gcManager, errCh: make(chan error, defaultErrChSize), cancel: func() {}, - - // newDDLPuller: newDDLPuller, - // newSink: newDDLSink, } c.newScheduler = newScheduler return c } -// TODO: modify for tikv cdc func newChangefeed4Test( id model.ChangeFeedID, gcManager gc.Manager, - /* - newDDLPuller func(ctx cdcContext.Context, startTs uint64) (DDLPuller, error), - newSink func() DDLSink, - */ ) *changefeed { c := newChangefeed(id, gcManager) - /* - c.newDDLPuller = newDDLPuller - c.newSink = newSink - */ + c.newScheduler = newScheduler4Test return c } @@ -239,61 +212,11 @@ LOOP: } } - /* - if c.state.Info.SyncPointEnabled { - c.barriers.Update(syncPointBarrier, checkpointTs) - } - */ - - // Since we are starting DDL puller from (checkpointTs-1) to make - // the DDL committed at checkpointTs executable by CDC, we need to set - // the DDL barrier to the correct start point. - - /* - c.barriers.Update(ddlJobBarrier, checkpointTs-1) - c.barriers.Update(finishBarrier, c.state.Info.GetTargetTs()) - */ - var err error // Note that (checkpointTs == ddl.FinishedTs) DOES NOT imply that the DDL has been completed executed. // So we need to process all DDLs from the range [checkpointTs, ...), but since the semantics of start-ts requires // the lower bound of an open interval, i.e. (startTs, ...), we pass checkpointTs-1 as the start-ts to initialize // the schema cache. - /* - c.schema, err = newSchemaWrap4Owner(ctx.GlobalVars().KVStorage, checkpointTs-1, c.state.Info.Config) - if err != nil { - return errors.Trace(err) - } - */ - - /* - cancelCtx, cancel := cdcContext.WithCancel(ctx) - c.cancel = cancel - - c.sink = c.newSink() - c.sink.run(cancelCtx, cancelCtx.ChangefeedVars().ID, cancelCtx.ChangefeedVars().Info) - */ - // Refer to the previous comment on why we use (checkpointTs-1). - - /* - c.ddlPuller, err = c.newDDLPuller(cancelCtx, checkpointTs-1) - if err != nil { - return errors.Trace(err) - } - c.wg.Add(1) - go func() { - defer c.wg.Done() - ctx.Throw(c.ddlPuller.Run(cancelCtx)) - }() - - stdCtx := util.PutChangefeedIDInCtx(cancelCtx, c.id) - redoManagerOpts := &redo.ManagerOptions{EnableBgRunner: false} - redoManager, err := redo.NewManager(stdCtx, c.state.Info.Config.Consistent, redoManagerOpts) - if err != nil { - return err - } - c.redoManager = redoManager - */ // init metrics c.metricsChangefeedCheckpointTsGauge = changefeedCheckpointTsGauge.WithLabelValues(c.id) @@ -313,27 +236,16 @@ LOOP: func (c *changefeed) releaseResources(ctx cdcContext.Context) { if !c.initialized { - // c.redoManagerCleanup(ctx) return } log.Info("close changefeed", zap.String("changefeed", c.state.ID), zap.Stringer("info", c.state.Info), zap.Bool("isRemoved", c.isRemoved)) c.cancel() c.cancel = func() {} - /* - c.schema = nil - c.ddlPuller.Close() - c.redoManagerCleanup(ctx) - */ _, cancel := context.WithCancel(context.Background()) cancel() // We don't need to wait sink Close, pass a canceled context is ok - /* - if err := c.sink.close(canceledCtx); err != nil { - log.Warn("Closing sink failed in Owner", zap.String("changefeedID", c.state.ID), zap.Error(err)) - } - */ c.wg.Wait() c.scheduler.Close(ctx) @@ -350,36 +262,6 @@ func (c *changefeed) releaseResources(ctx cdcContext.Context) { c.initialized = false } -/* -// redoManagerCleanup cleanups redo logs if changefeed is removed and redo log is enabled -func (c *changefeed) redoManagerCleanup(ctx context.Context) { - if c.isRemoved { - if c.state == nil || c.state.Info == nil || c.state.Info.Config == nil || - c.state.Info.Config.Consistent == nil { - log.Warn("changefeed is removed, but state is not complete", zap.Any("state", c.state)) - return - } - if !redo.IsConsistentEnabled(c.state.Info.Config.Consistent.Level) { - return - } - // when removing a paused changefeed, the redo manager is nil, create a new one - if c.redoManager == nil { - redoManagerOpts := &redo.ManagerOptions{EnableBgRunner: false} - redoManager, err := redo.NewManager(ctx, c.state.Info.Config.Consistent, redoManagerOpts) - if err != nil { - log.Error("create redo manager failed", zap.String("changefeed", c.id), zap.Error(err)) - return - } - c.redoManager = redoManager - } - err := c.redoManager.Cleanup(ctx) - if err != nil { - log.Error("cleanup redo logs failed", zap.String("changefeed", c.id), zap.Error(err)) - } - } -} -*/ - // preflightCheck makes sure that the metadata in Etcd is complete enough to run the tick. // If the metadata is not complete, such as when the ChangeFeedStatus is nil, // this function will reconstruct the lost metadata and skip this tick. @@ -440,101 +322,6 @@ func (c *changefeed) preflightCheck(captures map[model.CaptureID]*model.CaptureI return } -/* -func (c *changefeed) handleBarrier(ctx cdcContext.Context) (uint64, error) { - barrierTp, barrierTs := c.barriers.Min() - blocked := (barrierTs == c.state.Status.CheckpointTs) && (barrierTs == c.state.Status.ResolvedTs) - switch barrierTp { - case ddlJobBarrier: - ddlResolvedTs, ddlJob := c.ddlPuller.FrontDDL() - if ddlJob == nil || ddlResolvedTs != barrierTs { - if ddlResolvedTs < barrierTs { - return barrierTs, nil - } - c.barriers.Update(ddlJobBarrier, ddlResolvedTs) - return barrierTs, nil - } - if !blocked { - return barrierTs, nil - } - done, err := c.asyncExecDDL(ctx, ddlJob) - if err != nil { - return 0, errors.Trace(err) - } - if !done { - return barrierTs, nil - } - c.ddlPuller.PopFrontDDL() - newDDLResolvedTs, _ := c.ddlPuller.FrontDDL() - c.barriers.Update(ddlJobBarrier, newDDLResolvedTs) - - case syncPointBarrier: - if !blocked { - return barrierTs, nil - } - nextSyncPointTs := oracle.GoTimeToTS(oracle.GetTimeFromTS(barrierTs).Add(c.state.Info.SyncPointInterval)) - if err := c.sink.emitSyncPoint(ctx, barrierTs); err != nil { - return 0, errors.Trace(err) - } - c.barriers.Update(syncPointBarrier, nextSyncPointTs) - - case finishBarrier: - if !blocked { - return barrierTs, nil - } - c.feedStateManager.MarkFinished() - default: - log.Panic("Unknown barrier type", zap.Int("barrier type", int(barrierTp))) - } - return barrierTs, nil -} - -func (c *changefeed) asyncExecDDL(ctx cdcContext.Context, job *timodel.Job) (done bool, err error) { - if job.BinlogInfo == nil { - log.Warn("ignore the invalid DDL job", zap.Reflect("job", job)) - return true, nil - } - cyclicConfig := c.state.Info.Config.Cyclic - if cyclicConfig.IsEnabled() && !cyclicConfig.SyncDDL { - return true, nil - } - if c.ddlEventCache == nil || c.ddlEventCache.CommitTs != job.BinlogInfo.FinishedTS { - ddlEvent, err := c.schema.BuildDDLEvent(job) - if err != nil { - return false, errors.Trace(err) - } - err = c.schema.HandleDDL(job) - if err != nil { - return false, errors.Trace(err) - } - ddlEvent.Query, err = addSpecialComment(ddlEvent.Query) - if err != nil { - return false, errors.Trace(err) - } - - c.ddlEventCache = ddlEvent - if c.redoManager.Enabled() { - err = c.redoManager.EmitDDLEvent(ctx, ddlEvent) - if err != nil { - return false, err - } - } - } - if job.BinlogInfo.TableInfo != nil && c.schema.IsIneligibleTableID(job.BinlogInfo.TableInfo.ID) { - log.Warn("ignore the DDL job of ineligible table", zap.Reflect("job", job)) - return true, nil - } - done, err = c.sink.emitDDLEvent(ctx, c.ddlEventCache) - if err != nil { - return false, err - } - if done { - c.ddlEventCache = nil - } - return done, nil -} -*/ - func (c *changefeed) updateStatus(currentTs int64, checkpointTs, resolvedTs model.Ts) { c.state.PatchStatus(func(status *model.ChangeFeedStatus) (*model.ChangeFeedStatus, bool, error) { changed := false @@ -570,30 +357,3 @@ func (c *changefeed) GetInfoProvider() schedulerv2.InfoProvider { } return nil } - -/* -// addSpecialComment translate tidb feature to comment -// TODO: Maybe need delete this codes for tikv cdc. -func addSpecialComment(ddlQuery string) (string, error) { - stms, _, err := parser.New().ParseSQL(ddlQuery) - if err != nil { - return "", errors.Trace(err) - } - if len(stms) != 1 { - log.Panic("invalid ddlQuery statement size", zap.String("ddlQuery", ddlQuery)) - } - var sb strings.Builder - // translate TiDB feature to special comment - restoreFlags := format.RestoreTiDBSpecialComment - // escape the keyword - restoreFlags |= format.RestoreNameBackQuotes - // upper case keyword - restoreFlags |= format.RestoreKeyWordUppercase - // wrap string with single quote - restoreFlags |= format.RestoreStringSingleQuotes - if err = stms[0].Restore(format.NewRestoreCtx(restoreFlags, &sb)); err != nil { - return "", errors.Trace(err) - } - return sb.String(), nil -} -*/ diff --git a/cdc/cdc/owner/changefeed_test.go b/cdc/cdc/owner/changefeed_test.go index 290cd89c..28fd6777 100644 --- a/cdc/cdc/owner/changefeed_test.go +++ b/cdc/cdc/owner/changefeed_test.go @@ -15,7 +15,6 @@ package owner import ( "context" - "os" "path/filepath" "github.com/pingcap/check" @@ -105,6 +104,7 @@ func (s *changefeedSuite) TestHandleError(c *check.C) { ctx := cdcContext.NewBackendContext4Test(true) cf, state, captures, tester := createChangefeed4Test(ctx, c) defer cf.Close(ctx) + // pre check cf.Tick(ctx, state, captures) tester.MustApplyPatches() @@ -185,7 +185,4 @@ func testChangefeedReleaseResource( err := cf.tick(ctx, state, captures) c.Assert(err, check.IsNil) cancel() - // check redo log dir is deleted - _, err = os.Stat(redoLogDir) - c.Assert(os.IsNotExist(err), check.IsTrue) } diff --git a/cdc/cdc/owner/scheduler.go b/cdc/cdc/owner/scheduler.go index 21389a03..07c982c2 100644 --- a/cdc/cdc/owner/scheduler.go +++ b/cdc/cdc/owner/scheduler.go @@ -79,13 +79,14 @@ func NewSchedulerV2( checkpointTs model.Ts, messageServer *p2p.MessageServer, messageRouter p2p.MessageRouter, + f updateCurrentKeySpansFunc, ) (*schedulerV2, error) { ret := &schedulerV2{ changeFeedID: changeFeedID, messageServer: messageServer, messageRouter: messageRouter, stats: &schedulerStats{}, - updateCurrentKeySpans: updateCurrentKeySpansImpl, + updateCurrentKeySpans: f, } ret.BaseScheduleDispatcher = pscheduler.NewBaseScheduleDispatcher(changeFeedID, ret, checkpointTs) if err := ret.registerPeerMessageHandlers(ctx); err != nil { @@ -97,11 +98,11 @@ func NewSchedulerV2( // newSchedulerV2FromCtx creates a new schedulerV2 from context. // This function is factored out to facilitate unit testing. -func newSchedulerV2FromCtx(ctx context.Context, startTs uint64) (scheduler, error) { +func newSchedulerV2FromCtx(ctx context.Context, startTs uint64, f updateCurrentKeySpansFunc) (scheduler, error) { changeFeedID := ctx.ChangefeedVars().ID messageServer := ctx.GlobalVars().MessageServer messageRouter := ctx.GlobalVars().MessageRouter - ret, err := NewSchedulerV2(ctx, changeFeedID, startTs, messageServer, messageRouter) + ret, err := NewSchedulerV2(ctx, changeFeedID, startTs, messageServer, messageRouter, f) if err != nil { return nil, errors.Trace(err) } @@ -111,9 +112,17 @@ func newSchedulerV2FromCtx(ctx context.Context, startTs uint64) (scheduler, erro func newScheduler(ctx context.Context, startTs uint64) (scheduler, error) { conf := config.GetGlobalServerConfig() if conf.Debug.EnableNewScheduler { - return newSchedulerV2FromCtx(ctx, startTs) + return newSchedulerV2FromCtx(ctx, startTs, updateCurrentKeySpansImpl) } - return newSchedulerV1(), nil + return newSchedulerV1(updateCurrentKeySpansImpl), nil +} + +func newScheduler4Test(ctx context.Context, startTs uint64) (scheduler, error) { + conf := config.GetGlobalServerConfig() + if conf.Debug.EnableNewScheduler { + return newSchedulerV2FromCtx(ctx, startTs, updateCurrentKeySpansImpl4Test) + } + return newSchedulerV1(updateCurrentKeySpansImpl4Test), nil } func (s *schedulerV2) Tick( diff --git a/cdc/cdc/owner/scheduler_test.go b/cdc/cdc/owner/scheduler_test.go index 976adc96..f329c186 100644 --- a/cdc/cdc/owner/scheduler_test.go +++ b/cdc/cdc/owner/scheduler_test.go @@ -65,15 +65,7 @@ func TestSchedulerBasics(t *testing.T) { mockOwnerNode := mockCluster.Nodes["capture-0"] - sched, err := NewSchedulerV2( - ctx, - "cf-1", - 1000, - mockOwnerNode.Server, - mockOwnerNode.Router) - require.NoError(t, err) - - sched.updateCurrentKeySpans = func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { + f := func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { return []model.KeySpanID{1, 2, 3}, map[model.KeySpanID]regionspan.Span{ 1: {Start: []byte{'1'}, End: []byte{'2'}}, 2: {Start: []byte{'2'}, End: []byte{'3'}}, @@ -81,6 +73,16 @@ func TestSchedulerBasics(t *testing.T) { }, nil } + sched, err := NewSchedulerV2( + ctx, + "cf-1", + 1000, + mockOwnerNode.Server, + mockOwnerNode.Router, + f) + + require.NoError(t, err) + for atomic.LoadInt64(&sched.stats.AnnounceSentCount) < numNodes { checkpointTs, resolvedTs, err := sched.Tick(ctx, &orchestrator.ChangefeedReactorState{ ID: "cf-1", @@ -228,15 +230,7 @@ func TestSchedulerNoPeer(t *testing.T) { mockOwnerNode := mockCluster.Nodes["capture-0"] - sched, err := NewSchedulerV2( - ctx, - "cf-1", - 1000, - mockOwnerNode.Server, - mockOwnerNode.Router) - require.NoError(t, err) - - sched.updateCurrentKeySpans = func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { + f := func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { return []model.KeySpanID{1, 2, 3}, map[model.KeySpanID]regionspan.Span{ 1: {Start: []byte{'1'}, End: []byte{'2'}}, 2: {Start: []byte{'2'}, End: []byte{'3'}}, @@ -244,6 +238,15 @@ func TestSchedulerNoPeer(t *testing.T) { }, nil } + sched, err := NewSchedulerV2( + ctx, + "cf-1", + 1000, + mockOwnerNode.Server, + mockOwnerNode.Router, + f) + require.NoError(t, err) + // Ticks the scheduler 10 times. It should not panic. for i := 0; i < 10; i++ { checkpointTs, resolvedTs, err := sched.Tick(ctx, &orchestrator.ChangefeedReactorState{ diff --git a/cdc/cdc/owner/scheduler_v1.go b/cdc/cdc/owner/scheduler_v1.go index a5ebfcbd..705577bf 100644 --- a/cdc/cdc/owner/scheduler_v1.go +++ b/cdc/cdc/owner/scheduler_v1.go @@ -30,6 +30,7 @@ import ( ) type schedulerJobType string +type updateCurrentKeySpansFunc func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) const ( schedulerJobTypeAddKeySpan schedulerJobType = "ADD" @@ -65,13 +66,13 @@ type oldScheduler struct { needRebalanceNextTick bool lastTickCaptureCount int - updateCurrentKeySpans func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) + updateCurrentKeySpans updateCurrentKeySpansFunc } -func newSchedulerV1() scheduler { +func newSchedulerV1(f updateCurrentKeySpansFunc) scheduler { return &schedulerV1CompatWrapper{&oldScheduler{ moveKeySpanTargets: make(map[model.KeySpanID]model.CaptureID), - updateCurrentKeySpans: updateCurrentKeySpansImpl, + updateCurrentKeySpans: f, }} } @@ -81,11 +82,11 @@ func newSchedulerV1() scheduler { func (s *oldScheduler) Tick( ctx cdcContext.Context, state *orchestrator.ChangefeedReactorState, - // currentKeySpans []model.KeySpanID, captures map[model.CaptureID]*model.CaptureInfo, ) (shouldUpdateState bool, err error) { - s.state = state + s.captures = captures + currentKeySpanIDs, currentKeySpans, err := s.updateCurrentKeySpans(ctx) if err != nil { return false, errors.Trace(err) @@ -210,7 +211,7 @@ func (s *oldScheduler) keyspan2CaptureIndex() (map[model.KeySpanID]model.Capture // If the TargetCapture of a job is not set, it chooses a capture with the minimum workload(minimum number of keyspans) // and sets the TargetCapture to the capture. func (s *oldScheduler) dispatchToTargetCaptures(pendingJobs []*schedulerJob) { - workloads := make(map[model.CaptureID]uint64) + workloads := make(map[model.CaptureID]int64) for captureID := range s.captures { workloads[captureID] = 0 @@ -244,9 +245,11 @@ func (s *oldScheduler) dispatchToTargetCaptures(pendingJobs []*schedulerJob) { } } + count := 0 getMinWorkloadCapture := func() model.CaptureID { + count++ minCapture := "" - minWorkLoad := uint64(math.MaxUint64) + minWorkLoad := int64(math.MaxInt64) for captureID, workload := range workloads { if workload < minWorkLoad { minCapture = captureID @@ -491,6 +494,10 @@ func updateCurrentKeySpansImpl(ctx cdcContext.Context) ([]model.KeySpanID, map[m return currentKeySpansID, currentKeySpans, nil } +func updateCurrentKeySpansImpl4Test(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { + return nil, nil, nil +} + // schedulerV1CompatWrapper is used to wrap the old scheduler to // support the compatibility with the new scheduler. // It incorporates watermark calculations into the scheduler, which @@ -502,7 +509,6 @@ type schedulerV1CompatWrapper struct { func (w *schedulerV1CompatWrapper) Tick( ctx cdcContext.Context, state *orchestrator.ChangefeedReactorState, - // currentKeySpans []model.KeySpanID, captures map[model.CaptureID]*model.CaptureInfo, ) (newCheckpointTs, newResolvedTs model.Ts, err error) { @@ -548,6 +554,11 @@ func (w *schedulerV1CompatWrapper) calculateWatermarks( } } } + + if resolvedTs == model.Ts(math.MaxUint64) { + return schedulerv2.CheckpointCannotProceed, 0 + } + checkpointTs := resolvedTs for _, position := range state.TaskPositions { if checkpointTs > position.CheckPointTs { diff --git a/cdc/cdc/owner/scheduler_v1_test.go b/cdc/cdc/owner/scheduler_v1_test.go index aa7f425b..1ccd8c35 100644 --- a/cdc/cdc/owner/scheduler_v1_test.go +++ b/cdc/cdc/owner/scheduler_v1_test.go @@ -16,6 +16,7 @@ package owner import ( "fmt" "math/rand" + "testing" "github.com/pingcap/check" "github.com/tikv/migration/cdc/cdc/model" @@ -26,6 +27,10 @@ import ( "github.com/tikv/migration/cdc/pkg/util/testleak" ) +func Test(t *testing.T) { + check.TestingT(t) +} + var _ = check.Suite(&schedulerSuite{}) type schedulerSuite struct { @@ -40,7 +45,7 @@ func (s *schedulerSuite) reset(c *check.C) { s.changefeedID = fmt.Sprintf("test-changefeed-%x", rand.Uint32()) s.state = orchestrator.NewChangefeedReactorState("test-changefeed") s.tester = orchestrator.NewReactorStateTester(c, s.state, nil) - s.scheduler = newSchedulerV1().(*schedulerV1CompatWrapper).inner + s.scheduler = newSchedulerV1(updateCurrentKeySpansImpl4Test).(*schedulerV1CompatWrapper).inner s.captures = make(map[model.CaptureID]*model.CaptureInfo) s.state.PatchStatus(func(status *model.ChangeFeedStatus) (*model.ChangeFeedStatus, bool, error) { return &model.ChangeFeedStatus{}, true, nil @@ -95,6 +100,10 @@ func (s *schedulerSuite) TestScheduleOneCapture(c *check.C) { ctx, cancel := cdcContext.WithCancel(ctx) defer cancel() + s.scheduler.updateCurrentKeySpans = func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { + return nil, nil, nil + } + _, _ = s.scheduler.Tick(ctx, s.state, s.captures) // Manually simulate the scenario where the corresponding key was deleted in the etcd @@ -124,24 +133,25 @@ func (s *schedulerSuite) TestScheduleOneCapture(c *check.C) { c.Assert(shouldUpdateState, check.IsFalse) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 1: {StartTs: 0}, 2: {StartTs: 0}, 3: {StartTs: 0}, 4: {StartTs: 0}, + 1: {StartTs: 0, Start: []byte{'1'}, End: []byte{'2'}}, + 2: {StartTs: 0, Start: []byte{'2'}, End: []byte{'3'}}, + 3: {StartTs: 0, Start: []byte{'3'}, End: []byte{'4'}}, + 4: {StartTs: 0, Start: []byte{'4'}, End: []byte{'5'}}, }) c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ - 1: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, - 2: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, - 3: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, - 4: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, + 1: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{}}, + 2: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{}}, + 3: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{}}, + 4: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{}}, }) shouldUpdateState, err = s.scheduler.Tick(ctx, s.state, s.captures) // []model.KeySpanID{1, 2, 3, 4}, - c.Assert(err, check.IsNil) c.Assert(shouldUpdateState, check.IsTrue) s.tester.MustApplyPatches() // two keyspans finish adding operation s.finishKeySpanOperation(captureID, 2, 3) - s.scheduler.updateCurrentKeySpans = func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { return []model.KeySpanID{3, 4, 5}, map[model.KeySpanID]regionspan.Span{ 3: {Start: []byte{'3'}, End: []byte{'4'}}, @@ -155,13 +165,18 @@ func (s *schedulerSuite) TestScheduleOneCapture(c *check.C) { c.Assert(shouldUpdateState, check.IsFalse) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 3: {StartTs: 0}, 4: {StartTs: 0}, 5: {StartTs: 0}, + 3: {StartTs: 0, Start: []byte{'3'}, End: []byte{'4'}}, + 4: {StartTs: 0, Start: []byte{'4'}, End: []byte{'5'}}, + 5: {StartTs: 0, Start: []byte{'5'}, End: []byte{'6'}}, }) c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ - 1: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched}, - 2: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched}, - 4: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, - 5: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, + 1: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, + 2: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, + 4: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{}}, + 5: {Delete: false, + BoundaryTs: 0, + Status: model.OperDispatched, + RelatedKeySpans: []model.KeySpanLocation{{CaptureID: captureID, KeySpanID: 1}, {CaptureID: captureID, KeySpanID: 2}}}, }) // move a non exist keyspan to a non exist capture @@ -174,14 +189,18 @@ func (s *schedulerSuite) TestScheduleOneCapture(c *check.C) { c.Assert(shouldUpdateState, check.IsFalse) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 4: {StartTs: 0}, 5: {StartTs: 0}, + 4: {StartTs: 0, Start: []byte{'4'}, End: []byte{'5'}}, + 5: {StartTs: 0, Start: []byte{'5'}, End: []byte{'6'}}, }) c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ - 1: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched}, - 2: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched}, - 3: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched}, - 4: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, - 5: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, + 1: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, + 2: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, + 3: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, + 4: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{}}, + 5: {Delete: false, + BoundaryTs: 0, + Status: model.OperDispatched, + RelatedKeySpans: []model.KeySpanLocation{{CaptureID: captureID, KeySpanID: 1}, {CaptureID: captureID, KeySpanID: 2}}}, }) // finish all operations @@ -192,7 +211,8 @@ func (s *schedulerSuite) TestScheduleOneCapture(c *check.C) { c.Assert(shouldUpdateState, check.IsTrue) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 4: {StartTs: 0}, 5: {StartTs: 0}, + 4: {StartTs: 0, Start: []byte{'4'}, End: []byte{'5'}}, + 5: {StartTs: 0, Start: []byte{'5'}, End: []byte{'6'}}, }) c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{}) @@ -203,7 +223,8 @@ func (s *schedulerSuite) TestScheduleOneCapture(c *check.C) { c.Assert(shouldUpdateState, check.IsFalse) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 4: {StartTs: 0}, 5: {StartTs: 0}, + 4: {StartTs: 0, Start: []byte{'4'}, End: []byte{'5'}}, + 5: {StartTs: 0, Start: []byte{'5'}, End: []byte{'6'}}, }) c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{}) @@ -212,10 +233,12 @@ func (s *schedulerSuite) TestScheduleOneCapture(c *check.C) { c.Assert(shouldUpdateState, check.IsFalse) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 3: {StartTs: 0}, 4: {StartTs: 0}, 5: {StartTs: 0}, + 3: {StartTs: 0, Start: []byte{'3'}, End: []byte{'4'}}, + 4: {StartTs: 0, Start: []byte{'4'}, End: []byte{'5'}}, + 5: {StartTs: 0, Start: []byte{'5'}, End: []byte{'6'}}, }) c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ - 3: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, + 3: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, }) } @@ -241,10 +264,10 @@ func (s *schedulerSuite) TestScheduleMoveKeySpan(c *check.C) { c.Assert(shouldUpdateState, check.IsFalse) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID1].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 1: {StartTs: 0}, + 1: {StartTs: 0, Start: []byte{'1'}, End: []byte{'2'}}, }) c.Assert(s.state.TaskStatuses[captureID1].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ - 1: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, + 1: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{}}, }) s.finishKeySpanOperation(captureID1, 1) @@ -267,14 +290,14 @@ func (s *schedulerSuite) TestScheduleMoveKeySpan(c *check.C) { c.Assert(shouldUpdateState, check.IsFalse) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID1].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 1: {StartTs: 0}, + 1: {StartTs: 0, Start: []byte{'1'}, End: []byte{'2'}}, }) c.Assert(s.state.TaskStatuses[captureID1].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{}) c.Assert(s.state.TaskStatuses[captureID2].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 2: {StartTs: 0}, + 2: {StartTs: 0, Start: []byte{'2'}, End: []byte{'3'}}, }) c.Assert(s.state.TaskStatuses[captureID2].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ - 2: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, + 2: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{}}, }) s.finishKeySpanOperation(captureID2, 2) @@ -285,12 +308,12 @@ func (s *schedulerSuite) TestScheduleMoveKeySpan(c *check.C) { c.Assert(shouldUpdateState, check.IsFalse) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID1].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 1: {StartTs: 0}, + 1: {StartTs: 0, Start: []byte{'1'}, End: []byte{'2'}}, }) c.Assert(s.state.TaskStatuses[captureID1].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{}) c.Assert(s.state.TaskStatuses[captureID2].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{}) c.Assert(s.state.TaskStatuses[captureID2].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ - 2: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched}, + 2: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, }) s.finishKeySpanOperation(captureID2, 2) @@ -300,7 +323,7 @@ func (s *schedulerSuite) TestScheduleMoveKeySpan(c *check.C) { c.Assert(shouldUpdateState, check.IsTrue) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID1].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 1: {StartTs: 0}, + 1: {StartTs: 0, Start: []byte{'1'}, End: []byte{'2'}}, }) c.Assert(s.state.TaskStatuses[captureID1].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{}) c.Assert(s.state.TaskStatuses[captureID2].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{}) @@ -311,7 +334,8 @@ func (s *schedulerSuite) TestScheduleMoveKeySpan(c *check.C) { c.Assert(shouldUpdateState, check.IsFalse) s.tester.MustApplyPatches() c.Assert(s.state.TaskStatuses[captureID1].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ - 1: {StartTs: 0}, 2: {StartTs: 0}, + 1: {StartTs: 0, Start: []byte{'1'}, End: []byte{'2'}}, + 2: {StartTs: 0, Start: []byte{'2'}, End: []byte{'3'}}, }) c.Assert(s.state.TaskStatuses[captureID1].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ 2: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched}, @@ -407,3 +431,79 @@ func (s *schedulerSuite) TestScheduleRebalance(c *check.C) { } c.Assert(keyspanIDs, check.DeepEquals, map[model.KeySpanID]struct{}{1: {}, 2: {}, 3: {}, 4: {}, 5: {}, 6: {}}) } + +func (s *schedulerSuite) TestRelatedKeySpans(c *check.C) { + defer testleak.AfterTest(c)() + s.reset(c) + captureID := "test-capture" + s.addCapture(captureID) + + ctx := cdcContext.NewBackendContext4Test(false) + ctx, cancel := cdcContext.WithCancel(ctx) + defer cancel() + + s.scheduler.updateCurrentKeySpans = func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { + return []model.KeySpanID{1}, map[model.KeySpanID]regionspan.Span{ + 1: {Start: []byte{'1'}, End: []byte{'3'}}, + }, nil + } + + shouldUpdateState, err := s.scheduler.Tick(ctx, s.state, s.captures) // []model.KeySpanID{1}, + c.Assert(err, check.IsNil) + c.Assert(shouldUpdateState, check.IsFalse) + s.tester.MustApplyPatches() + c.Assert(s.state.TaskStatuses[captureID].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ + 1: {StartTs: 0, Start: []byte{'1'}, End: []byte{'3'}}, + }) + c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ + 1: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{}}, + }) + + s.scheduler.updateCurrentKeySpans = func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { + return []model.KeySpanID{2, 3}, map[model.KeySpanID]regionspan.Span{ + 2: {Start: []byte{'1'}, End: []byte{'2'}}, 3: {Start: []byte{'2'}, End: []byte{'3'}}, + }, nil + } + + shouldUpdateState, err = s.scheduler.Tick(ctx, s.state, s.captures) // []model.KeySpanID{2, 3}, + c.Assert(err, check.IsNil) + c.Assert(shouldUpdateState, check.IsFalse) + s.tester.MustApplyPatches() + c.Assert(s.state.TaskStatuses[captureID].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ + 2: {StartTs: 0, Start: []byte{'1'}, End: []byte{'2'}}, + 3: {StartTs: 0, Start: []byte{'2'}, End: []byte{'3'}}, + }) + c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ + 1: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, + 2: {Delete: false, + BoundaryTs: 0, + Status: model.OperDispatched, + RelatedKeySpans: []model.KeySpanLocation{{CaptureID: captureID, KeySpanID: 1}}}, + 3: {Delete: false, + BoundaryTs: 0, + Status: model.OperDispatched, + RelatedKeySpans: []model.KeySpanLocation{{CaptureID: captureID, KeySpanID: 1}}}, + }) + + s.scheduler.updateCurrentKeySpans = func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { + return []model.KeySpanID{4}, map[model.KeySpanID]regionspan.Span{ + 4: {Start: []byte{'1'}, End: []byte{'3'}}, + }, nil + } + shouldUpdateState, err = s.scheduler.Tick(ctx, s.state, s.captures) // []model.KeySpanID{4}, + c.Assert(err, check.IsNil) + c.Assert(shouldUpdateState, check.IsFalse) + s.tester.MustApplyPatches() + c.Assert(s.state.TaskStatuses[captureID].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ + 4: {StartTs: 0, Start: []byte{'1'}, End: []byte{'3'}}, + }) + c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ + 1: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, + 2: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, + 3: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, + 4: {Delete: false, + BoundaryTs: 0, + Status: model.OperDispatched, + RelatedKeySpans: []model.KeySpanLocation{{CaptureID: captureID, KeySpanID: 2}, {CaptureID: captureID, KeySpanID: 3}}}, + }) +} From edbfecdc0bfc132b44ed5f7d29a1d1c8ba98a46b Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Tue, 10 May 2022 16:47:56 +0800 Subject: [PATCH 07/13] add ut for processor Signed-off-by: zeminzhou --- cdc/cdc/processor/processor_test.go | 77 +++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/cdc/cdc/processor/processor_test.go b/cdc/cdc/processor/processor_test.go index 0b1a21f3..faeab9c2 100644 --- a/cdc/cdc/processor/processor_test.go +++ b/cdc/cdc/processor/processor_test.go @@ -926,3 +926,80 @@ func (s *processorSuite) TestIgnorableError(c *check.C) { c.Assert(isProcessorIgnorableError(tc.err), check.Equals, tc.ignorable) } } + +func (s *processorSuite) TestHandleKeySpanOperationWithRelatedKeySpans(c *check.C) { + defer testleak.AfterTest(c)() + ctx := cdcContext.NewBackendContext4Test(true) + p, tester := initProcessor4Test(ctx, c) + var err error + + // no operation + _, err = p.Tick(ctx, p.changefeed) + c.Assert(err, check.IsNil) + tester.MustApplyPatches() + + // add keyspan1, in processing + p.changefeed.PatchTaskStatus(p.captureInfo.ID, func(status *model.TaskStatus) (*model.TaskStatus, bool, error) { + status.AddKeySpan(1, &model.KeySpanReplicaInfo{StartTs: 60}, 80, nil) + return status, true, nil + }) + tester.MustApplyPatches() + _, err = p.Tick(ctx, p.changefeed) + c.Assert(err, check.IsNil) + tester.MustApplyPatches() + c.Assert(p.changefeed.TaskStatuses[p.captureInfo.ID], check.DeepEquals, &model.TaskStatus{ + KeySpans: map[uint64]*model.KeySpanReplicaInfo{ + 1: {StartTs: 60}, + }, + Operation: map[uint64]*model.KeySpanOperation{ + 1: {Delete: false, BoundaryTs: 80, Status: model.OperProcessed}, + }, + }) + c.Assert(p.keyspans, check.HasLen, 1) + c.Assert(p.changefeed.TaskPositions[p.captureInfo.ID].CheckPointTs, check.Equals, uint64(60)) + c.Assert(p.changefeed.TaskPositions[p.captureInfo.ID].ResolvedTs, check.Equals, uint64(60)) + + // add keyspan2 & keyspan3, remove keyspan1 + p.changefeed.PatchTaskStatus(p.captureInfo.ID, func(status *model.TaskStatus) (*model.TaskStatus, bool, error) { + status.AddKeySpan(2, &model.KeySpanReplicaInfo{StartTs: 60}, 60, []model.KeySpanLocation{{CaptureID: p.captureInfo.ID, KeySpanID: 1}}) + status.AddKeySpan(3, &model.KeySpanReplicaInfo{StartTs: 60}, 60, []model.KeySpanLocation{{CaptureID: p.captureInfo.ID, KeySpanID: 1}}) + status.RemoveKeySpan(1, 60, false) + return status, true, nil + }) + tester.MustApplyPatches() + // try to stop keyspand1 + _, err = p.Tick(ctx, p.changefeed) + c.Assert(err, check.IsNil) + tester.MustApplyPatches() + c.Assert(p.changefeed.TaskStatuses[p.captureInfo.ID].KeySpans, check.DeepEquals, map[uint64]*model.KeySpanReplicaInfo{ + 2: {StartTs: 60}, + 3: {StartTs: 60}, + }) + c.Assert(p.changefeed.TaskStatuses[p.captureInfo.ID].Operation, check.DeepEquals, map[uint64]*model.KeySpanOperation{ + 1: {Delete: true, BoundaryTs: 60, Status: model.OperProcessed}, + 2: {Delete: false, BoundaryTs: 60, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{{CaptureID: p.captureInfo.ID, KeySpanID: 1}}}, + 3: {Delete: false, BoundaryTs: 60, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{{CaptureID: p.captureInfo.ID, KeySpanID: 1}}}, + }) + keyspan1 := p.keyspans[1].(*mockKeySpanPipeline) + keyspan1.status = keyspanpipeline.KeySpanStatusStopped + + // finish stoping keyspand1 + _, err = p.Tick(ctx, p.changefeed) + c.Assert(err, check.IsNil) + tester.MustApplyPatches() + c.Assert(p.changefeed.TaskStatuses[p.captureInfo.ID].Operation, check.DeepEquals, map[uint64]*model.KeySpanOperation{ + 1: {Delete: true, BoundaryTs: 60, Status: model.OperFinished}, + 2: {Delete: false, BoundaryTs: 60, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{{CaptureID: p.captureInfo.ID, KeySpanID: 1}}}, + 3: {Delete: false, BoundaryTs: 60, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{{CaptureID: p.captureInfo.ID, KeySpanID: 1}}}, + }) + cleanUpFinishedOpOperation(p.changefeed, p.captureInfo.ID, tester) + + // start keyspan2 & keyspan3 + _, err = p.Tick(ctx, p.changefeed) + c.Assert(err, check.IsNil) + tester.MustApplyPatches() + c.Assert(p.changefeed.TaskStatuses[p.captureInfo.ID].Operation, check.DeepEquals, map[uint64]*model.KeySpanOperation{ + 2: {Delete: false, BoundaryTs: 60, Status: model.OperProcessed, RelatedKeySpans: []model.KeySpanLocation{{CaptureID: p.captureInfo.ID, KeySpanID: 1}}}, + 3: {Delete: false, BoundaryTs: 60, Status: model.OperProcessed, RelatedKeySpans: []model.KeySpanLocation{{CaptureID: p.captureInfo.ID, KeySpanID: 1}}}, + }) +} From 401e190652866895f8ece88a9e76eb822d52cef4 Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Tue, 10 May 2022 17:34:58 +0800 Subject: [PATCH 08/13] fix ut Signed-off-by: zeminzhou --- cdc/cdc/model/owner_test.go | 4 ++-- cdc/cdc/owner/changefeed_test.go | 5 ++--- cdc/pkg/scheduler/workload.go | 18 +++++++++--------- cdc/pkg/scheduler/workload_test.go | 2 +- 4 files changed, 14 insertions(+), 15 deletions(-) diff --git a/cdc/cdc/model/owner_test.go b/cdc/cdc/model/owner_test.go index 77d303ac..f73e5b6f 100644 --- a/cdc/cdc/model/owner_test.go +++ b/cdc/cdc/model/owner_test.go @@ -120,8 +120,8 @@ func TestTaskWorkloadMarshal(t *testing.T) { t.Parallel() workload := &TaskWorkload{ - 12: WorkloadInfo{Workload: uint64(1)}, - 15: WorkloadInfo{Workload: uint64(3)}, + 12: WorkloadInfo{Workload: int64(1)}, + 15: WorkloadInfo{Workload: int64(3)}, } expected := `{"12":{"workload":1},"15":{"workload":3}}` diff --git a/cdc/cdc/owner/changefeed_test.go b/cdc/cdc/owner/changefeed_test.go index 28fd6777..3cefcda1 100644 --- a/cdc/cdc/owner/changefeed_test.go +++ b/cdc/cdc/owner/changefeed_test.go @@ -136,7 +136,7 @@ func (s *changefeedSuite) TestRemoveChangefeed(c *check.C) { ID: ctx.ChangefeedVars().ID, Info: info, }) - testChangefeedReleaseResource(c, ctx, cancel, dir, true /*expectedInitialized*/) + testChangefeedReleaseResource(c, ctx, cancel, true /*expectedInitialized*/) } func (s *changefeedSuite) TestRemovePausedChangefeed(c *check.C) { @@ -155,14 +155,13 @@ func (s *changefeedSuite) TestRemovePausedChangefeed(c *check.C) { ID: ctx.ChangefeedVars().ID, Info: info, }) - testChangefeedReleaseResource(c, ctx, cancel, dir, false /*expectedInitialized*/) + testChangefeedReleaseResource(c, ctx, cancel, false /*expectedInitialized*/) } func testChangefeedReleaseResource( c *check.C, ctx cdcContext.Context, cancel context.CancelFunc, - redoLogDir string, expectedInitialized bool, ) { cf, state, captures, tester := createChangefeed4Test(ctx, c) diff --git a/cdc/pkg/scheduler/workload.go b/cdc/pkg/scheduler/workload.go index f33f2ca7..d4d89a14 100644 --- a/cdc/pkg/scheduler/workload.go +++ b/cdc/pkg/scheduler/workload.go @@ -55,23 +55,23 @@ func (w workloads) RemoveKeySpan(captureID model.CaptureID, keyspanID model.KeyS delete(captureWorkloads, keyspanID) } -func (w workloads) AvgEachKeySpan() uint64 { - var totalWorkload uint64 - var totalKeySpan uint64 +func (w workloads) AvgEachKeySpan() int64 { + var totalWorkload int64 + var totalKeySpan int64 for _, captureWorkloads := range w { for _, workload := range captureWorkloads { totalWorkload += workload.Workload } - totalKeySpan += uint64(len(captureWorkloads)) + totalKeySpan += int64(len(captureWorkloads)) } return totalWorkload / totalKeySpan } func (w workloads) Skewness() float64 { - totalWorkloads := make([]uint64, 0, len(w)) - var workloadSum uint64 + totalWorkloads := make([]int64, 0, len(w)) + var workloadSum int64 for _, captureWorkloads := range w { - var total uint64 + var total int64 for _, workload := range captureWorkloads { total += workload.Workload } @@ -88,10 +88,10 @@ func (w workloads) Skewness() float64 { } func (w workloads) SelectIdleCapture() model.CaptureID { - minWorkload := uint64(math.MaxUint64) + minWorkload := int64(math.MaxInt64) var minCapture model.CaptureID for captureID, captureWorkloads := range w { - var totalWorkloadInCapture uint64 + var totalWorkloadInCapture int64 for _, workload := range captureWorkloads { totalWorkloadInCapture += workload.Workload } diff --git a/cdc/pkg/scheduler/workload_test.go b/cdc/pkg/scheduler/workload_test.go index 4869ad03..6931ce4d 100644 --- a/cdc/pkg/scheduler/workload_test.go +++ b/cdc/pkg/scheduler/workload_test.go @@ -43,7 +43,7 @@ func TestWorkloads(t *testing.T) { "capture2": {4: model.WorkloadInfo{Workload: 1}, 3: model.WorkloadInfo{Workload: 2}, 5: model.WorkloadInfo{Workload: 8}}, "capture3": {6: model.WorkloadInfo{Workload: 1}}, }) - require.Equal(t, w.AvgEachKeySpan(), uint64(2+1+2+8+1)/5) + require.Equal(t, w.AvgEachKeySpan(), int64(2+1+2+8+1)/5) require.Equal(t, w.SelectIdleCapture(), "capture3") require.Equal(t, fmt.Sprintf("%.2f%%", w.Skewness()*100), "96.36%") From 981497b1f2f29c1ec2f8e030e03cd583510afc2d Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Tue, 14 Jun 2022 16:54:55 +0800 Subject: [PATCH 09/13] fix ut Signed-off-by: zeminzhou --- cdc/cdc/owner/scheduler_v1_test.go | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/cdc/cdc/owner/scheduler_v1_test.go b/cdc/cdc/owner/scheduler_v1_test.go index 98c92f74..daf44a5f 100644 --- a/cdc/cdc/owner/scheduler_v1_test.go +++ b/cdc/cdc/owner/scheduler_v1_test.go @@ -16,6 +16,7 @@ package owner import ( "fmt" "math/rand" + "sort" "testing" "github.com/pingcap/check" @@ -171,6 +172,12 @@ func (s *schedulerSuite) TestScheduleOneCapture(c *check.C) { 4: {StartTs: 0, Start: []byte{'4'}, End: []byte{'5'}}, 5: {StartTs: 0, Start: []byte{'5'}, End: []byte{'6'}}, }) + + keyspanOperation, IsTrue := s.state.TaskStatuses[captureID].Operation[5] + c.Assert(IsTrue, check.IsTrue) + sort.SliceStable(keyspanOperation.RelatedKeySpans, func(i, j int) bool { + return keyspanOperation.RelatedKeySpans[i].KeySpanID < keyspanOperation.RelatedKeySpans[j].KeySpanID + }) c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ 1: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, 2: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, @@ -194,6 +201,12 @@ func (s *schedulerSuite) TestScheduleOneCapture(c *check.C) { 4: {StartTs: 0, Start: []byte{'4'}, End: []byte{'5'}}, 5: {StartTs: 0, Start: []byte{'5'}, End: []byte{'6'}}, }) + + keyspanOperation, IsTrue = s.state.TaskStatuses[captureID].Operation[5] + c.Assert(IsTrue, check.IsTrue) + sort.SliceStable(keyspanOperation.RelatedKeySpans, func(i, j int) bool { + return keyspanOperation.RelatedKeySpans[i].KeySpanID < keyspanOperation.RelatedKeySpans[j].KeySpanID + }) c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ 1: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, 2: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, @@ -499,6 +512,13 @@ func (s *schedulerSuite) TestRelatedKeySpans(c *check.C) { c.Assert(s.state.TaskStatuses[captureID].KeySpans, check.DeepEquals, map[model.KeySpanID]*model.KeySpanReplicaInfo{ 4: {StartTs: 0, Start: []byte{'1'}, End: []byte{'3'}}, }) + + keyspanOperation, IsTrue := s.state.TaskStatuses[captureID].Operation[4] + c.Assert(IsTrue, check.IsTrue) + sort.SliceStable(keyspanOperation.RelatedKeySpans, func(i, j int) bool { + return keyspanOperation.RelatedKeySpans[i].KeySpanID < keyspanOperation.RelatedKeySpans[j].KeySpanID + }) + c.Assert(s.state.TaskStatuses[captureID].Operation, check.DeepEquals, map[model.KeySpanID]*model.KeySpanOperation{ 1: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, 2: {Delete: true, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: nil}, From fb38d21f825b957723bab00cb22a71f99943abad Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Tue, 14 Jun 2022 18:16:19 +0800 Subject: [PATCH 10/13] remove spaceline Signed-off-by: zeminzhou --- cdc/cdc/owner/changefeed.go | 1 - cdc/cdc/owner/changefeed_test.go | 1 - cdc/cdc/owner/scheduler_test.go | 1 - 3 files changed, 3 deletions(-) diff --git a/cdc/cdc/owner/changefeed.go b/cdc/cdc/owner/changefeed.go index 16ef02ce..78ec4b25 100644 --- a/cdc/cdc/owner/changefeed.go +++ b/cdc/cdc/owner/changefeed.go @@ -44,7 +44,6 @@ type changefeed struct { feedStateManager *feedStateManager gcManager gc.Manager initialized bool - // isRemoved is true if the changefeed is removed isRemoved bool diff --git a/cdc/cdc/owner/changefeed_test.go b/cdc/cdc/owner/changefeed_test.go index 3cefcda1..607226a2 100644 --- a/cdc/cdc/owner/changefeed_test.go +++ b/cdc/cdc/owner/changefeed_test.go @@ -104,7 +104,6 @@ func (s *changefeedSuite) TestHandleError(c *check.C) { ctx := cdcContext.NewBackendContext4Test(true) cf, state, captures, tester := createChangefeed4Test(ctx, c) defer cf.Close(ctx) - // pre check cf.Tick(ctx, state, captures) tester.MustApplyPatches() diff --git a/cdc/cdc/owner/scheduler_test.go b/cdc/cdc/owner/scheduler_test.go index f329c186..f25888ac 100644 --- a/cdc/cdc/owner/scheduler_test.go +++ b/cdc/cdc/owner/scheduler_test.go @@ -80,7 +80,6 @@ func TestSchedulerBasics(t *testing.T) { mockOwnerNode.Server, mockOwnerNode.Router, f) - require.NoError(t, err) for atomic.LoadInt64(&sched.stats.AnnounceSentCount) < numNodes { From 272db7668607209882bed82d67b38379ad5421b5 Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Tue, 14 Jun 2022 21:12:52 +0800 Subject: [PATCH 11/13] int64 -> uint64 Signed-off-by: zeminzhou --- cdc/cdc/model/owner.go | 2 +- cdc/cdc/model/owner_test.go | 4 +-- cdc/cdc/owner/scheduler_v1.go | 4 +-- cdc/cdc/owner/scheduler_v1_test.go | 51 ++++++++++++++++++++++++++++++ cdc/pkg/scheduler/workload.go | 18 +++++------ cdc/pkg/scheduler/workload_test.go | 2 +- 6 files changed, 66 insertions(+), 15 deletions(-) diff --git a/cdc/cdc/model/owner.go b/cdc/cdc/model/owner.go index 958c8398..26e35136 100644 --- a/cdc/cdc/model/owner.go +++ b/cdc/cdc/model/owner.go @@ -195,7 +195,7 @@ type TaskWorkload map[KeySpanID]WorkloadInfo // WorkloadInfo records the workload info of a keyspan type WorkloadInfo struct { - Workload int64 `json:"workload"` + Workload uint64 `json:"workload"` } // Unmarshal unmarshals into *TaskWorkload from json marshal byte slice diff --git a/cdc/cdc/model/owner_test.go b/cdc/cdc/model/owner_test.go index f73e5b6f..77d303ac 100644 --- a/cdc/cdc/model/owner_test.go +++ b/cdc/cdc/model/owner_test.go @@ -120,8 +120,8 @@ func TestTaskWorkloadMarshal(t *testing.T) { t.Parallel() workload := &TaskWorkload{ - 12: WorkloadInfo{Workload: int64(1)}, - 15: WorkloadInfo{Workload: int64(3)}, + 12: WorkloadInfo{Workload: uint64(1)}, + 15: WorkloadInfo{Workload: uint64(3)}, } expected := `{"12":{"workload":1},"15":{"workload":3}}` diff --git a/cdc/cdc/owner/scheduler_v1.go b/cdc/cdc/owner/scheduler_v1.go index 705577bf..cde323ea 100644 --- a/cdc/cdc/owner/scheduler_v1.go +++ b/cdc/cdc/owner/scheduler_v1.go @@ -211,7 +211,7 @@ func (s *oldScheduler) keyspan2CaptureIndex() (map[model.KeySpanID]model.Capture // If the TargetCapture of a job is not set, it chooses a capture with the minimum workload(minimum number of keyspans) // and sets the TargetCapture to the capture. func (s *oldScheduler) dispatchToTargetCaptures(pendingJobs []*schedulerJob) { - workloads := make(map[model.CaptureID]int64) + workloads := make(map[model.CaptureID]uint64) for captureID := range s.captures { workloads[captureID] = 0 @@ -249,7 +249,7 @@ func (s *oldScheduler) dispatchToTargetCaptures(pendingJobs []*schedulerJob) { getMinWorkloadCapture := func() model.CaptureID { count++ minCapture := "" - minWorkLoad := int64(math.MaxInt64) + minWorkLoad := uint64(math.MaxUint64) for captureID, workload := range workloads { if workload < minWorkLoad { minCapture = captureID diff --git a/cdc/cdc/owner/scheduler_v1_test.go b/cdc/cdc/owner/scheduler_v1_test.go index daf44a5f..22de1bbf 100644 --- a/cdc/cdc/owner/scheduler_v1_test.go +++ b/cdc/cdc/owner/scheduler_v1_test.go @@ -474,6 +474,23 @@ func (s *schedulerSuite) TestRelatedKeySpans(c *check.C) { 1: {Delete: false, BoundaryTs: 0, Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{}}, }) + s.state.PatchTaskWorkload(captureID, func(workload model.TaskWorkload) (model.TaskWorkload, bool, error) { + if workload == nil { + workload = make(model.TaskWorkload) + } + for keyspanID := range s.state.TaskStatuses[captureID].KeySpans { + if s.state.TaskStatuses[captureID].Operation[keyspanID].Delete { + delete(workload, keyspanID) + } else { + workload[keyspanID] = model.WorkloadInfo{ + Workload: 1, + } + } + } + return workload, true, nil + }) + s.tester.MustApplyPatches() + s.scheduler.updateCurrentKeySpans = func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { return []model.KeySpanID{2, 3}, map[model.KeySpanID]regionspan.Span{ 2: {Start: []byte{'1'}, End: []byte{'2'}}, 3: {Start: []byte{'2'}, End: []byte{'3'}}, @@ -500,6 +517,23 @@ func (s *schedulerSuite) TestRelatedKeySpans(c *check.C) { RelatedKeySpans: []model.KeySpanLocation{{CaptureID: captureID, KeySpanID: 1}}}, }) + s.state.PatchTaskWorkload(captureID, func(workload model.TaskWorkload) (model.TaskWorkload, bool, error) { + if workload == nil { + workload = make(model.TaskWorkload) + } + for keyspanID := range s.state.TaskStatuses[captureID].KeySpans { + if s.state.TaskStatuses[captureID].Operation[keyspanID].Delete { + delete(workload, keyspanID) + } else { + workload[keyspanID] = model.WorkloadInfo{ + Workload: 1, + } + } + } + return workload, true, nil + }) + s.tester.MustApplyPatches() + s.scheduler.updateCurrentKeySpans = func(ctx cdcContext.Context) ([]model.KeySpanID, map[model.KeySpanID]regionspan.Span, error) { return []model.KeySpanID{4}, map[model.KeySpanID]regionspan.Span{ 4: {Start: []byte{'1'}, End: []byte{'3'}}, @@ -528,4 +562,21 @@ func (s *schedulerSuite) TestRelatedKeySpans(c *check.C) { Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{{CaptureID: captureID, KeySpanID: 2}, {CaptureID: captureID, KeySpanID: 3}}}, }) + + s.state.PatchTaskWorkload(captureID, func(workload model.TaskWorkload) (model.TaskWorkload, bool, error) { + if workload == nil { + workload = make(model.TaskWorkload) + } + for keyspanID := range s.state.TaskStatuses[captureID].KeySpans { + if s.state.TaskStatuses[captureID].Operation[keyspanID].Delete { + delete(workload, keyspanID) + } else { + workload[keyspanID] = model.WorkloadInfo{ + Workload: 1, + } + } + } + return workload, true, nil + }) + s.tester.MustApplyPatches() } diff --git a/cdc/pkg/scheduler/workload.go b/cdc/pkg/scheduler/workload.go index d4d89a14..d2597dd2 100644 --- a/cdc/pkg/scheduler/workload.go +++ b/cdc/pkg/scheduler/workload.go @@ -55,23 +55,23 @@ func (w workloads) RemoveKeySpan(captureID model.CaptureID, keyspanID model.KeyS delete(captureWorkloads, keyspanID) } -func (w workloads) AvgEachKeySpan() int64 { - var totalWorkload int64 - var totalKeySpan int64 +func (w workloads) AvgEachKeySpan() uint64 { + var totalWorkload uint64 + var totalKeySpan uint64 for _, captureWorkloads := range w { for _, workload := range captureWorkloads { totalWorkload += workload.Workload } - totalKeySpan += int64(len(captureWorkloads)) + totalKeySpan += uint64(len(captureWorkloads)) } return totalWorkload / totalKeySpan } func (w workloads) Skewness() float64 { - totalWorkloads := make([]int64, 0, len(w)) - var workloadSum int64 + totalWorkloads := make([]uint64, 0, len(w)) + var workloadSum uint64 for _, captureWorkloads := range w { - var total int64 + var total uint64 for _, workload := range captureWorkloads { total += workload.Workload } @@ -88,10 +88,10 @@ func (w workloads) Skewness() float64 { } func (w workloads) SelectIdleCapture() model.CaptureID { - minWorkload := int64(math.MaxInt64) + minWorkload := uint64(math.MaxInt64) var minCapture model.CaptureID for captureID, captureWorkloads := range w { - var totalWorkloadInCapture int64 + var totalWorkloadInCapture uint64 for _, workload := range captureWorkloads { totalWorkloadInCapture += workload.Workload } diff --git a/cdc/pkg/scheduler/workload_test.go b/cdc/pkg/scheduler/workload_test.go index 6931ce4d..4869ad03 100644 --- a/cdc/pkg/scheduler/workload_test.go +++ b/cdc/pkg/scheduler/workload_test.go @@ -43,7 +43,7 @@ func TestWorkloads(t *testing.T) { "capture2": {4: model.WorkloadInfo{Workload: 1}, 3: model.WorkloadInfo{Workload: 2}, 5: model.WorkloadInfo{Workload: 8}}, "capture3": {6: model.WorkloadInfo{Workload: 1}}, }) - require.Equal(t, w.AvgEachKeySpan(), int64(2+1+2+8+1)/5) + require.Equal(t, w.AvgEachKeySpan(), uint64(2+1+2+8+1)/5) require.Equal(t, w.SelectIdleCapture(), "capture3") require.Equal(t, fmt.Sprintf("%.2f%%", w.Skewness()*100), "96.36%") From 176b48ec041c550621dbc784efe04d30cc576037 Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Tue, 14 Jun 2022 23:07:21 +0800 Subject: [PATCH 12/13] . Signed-off-by: zeminzhou --- cdc/cdc/owner/scheduler_v1_test.go | 17 ----------------- 1 file changed, 17 deletions(-) diff --git a/cdc/cdc/owner/scheduler_v1_test.go b/cdc/cdc/owner/scheduler_v1_test.go index 22de1bbf..7fdbc651 100644 --- a/cdc/cdc/owner/scheduler_v1_test.go +++ b/cdc/cdc/owner/scheduler_v1_test.go @@ -562,21 +562,4 @@ func (s *schedulerSuite) TestRelatedKeySpans(c *check.C) { Status: model.OperDispatched, RelatedKeySpans: []model.KeySpanLocation{{CaptureID: captureID, KeySpanID: 2}, {CaptureID: captureID, KeySpanID: 3}}}, }) - - s.state.PatchTaskWorkload(captureID, func(workload model.TaskWorkload) (model.TaskWorkload, bool, error) { - if workload == nil { - workload = make(model.TaskWorkload) - } - for keyspanID := range s.state.TaskStatuses[captureID].KeySpans { - if s.state.TaskStatuses[captureID].Operation[keyspanID].Delete { - delete(workload, keyspanID) - } else { - workload[keyspanID] = model.WorkloadInfo{ - Workload: 1, - } - } - } - return workload, true, nil - }) - s.tester.MustApplyPatches() } From ea39b9d2963bb3bb0d482ed252d36bf3a87a4751 Mon Sep 17 00:00:00 2001 From: zeminzhou Date: Wed, 15 Jun 2022 09:55:10 +0800 Subject: [PATCH 13/13] . Signed-off-by: zeminzhou --- cdc/pkg/scheduler/workload.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cdc/pkg/scheduler/workload.go b/cdc/pkg/scheduler/workload.go index d2597dd2..f33f2ca7 100644 --- a/cdc/pkg/scheduler/workload.go +++ b/cdc/pkg/scheduler/workload.go @@ -88,7 +88,7 @@ func (w workloads) Skewness() float64 { } func (w workloads) SelectIdleCapture() model.CaptureID { - minWorkload := uint64(math.MaxInt64) + minWorkload := uint64(math.MaxUint64) var minCapture model.CaptureID for captureID, captureWorkloads := range w { var totalWorkloadInCapture uint64