support to balance subtasks to new captures

cdc/changefeed.go

@@ -429,9 +429,10 @@ func (c *changeFeed) rebanlanceTables(ctx context.Context, captures map[model.Ca
 			return nil
 		}
 	}
+	timeToRebanlance := time.Since(c.lastRebanlanceTime) > time.Duration(c.info.Config.Scheduler.PollingTime)*time.Minute
+	timeToRebanlance = timeToRebanlance && c.info.Config.Scheduler.PollingTime > 0
 
-	if !c.rebanlanceNextTick &&
-		time.Since(c.lastRebanlanceTime) < time.Duration(c.info.Config.Scheduler.PollingTime)*time.Minute {
+	if !c.rebanlanceNextTick && !timeToRebanlance {
 		return nil
 	}
 	c.lastRebanlanceTime = time.Now()

cdc/processor.go

@@ -397,7 +397,7 @@ func (p *processor) ddlPullWorker(ctx context.Context) error {
 }
 
 func (p *processor) workloadWorker(ctx context.Context) error {
-	t := time.NewTicker(1 * time.Minute)
+	t := time.NewTicker(10 * time.Second)
 	err := p.etcdCli.PutTaskWorkload(ctx, p.changefeedID, p.captureID, nil)
 	if err != nil {
 		return errors.Trace(err)

pkg/config/config.go

@@ -39,8 +39,8 @@ var defaultReplicaConfig = &ReplicaConfig{
 		Enable: false,
 	},
 	Scheduler: &SchedulerConfig{
-		Tp:          "manual",
-		PollingTime: 5,
+		Tp:          "table-number",
+		PollingTime: -1,
 	},
 }
 

pkg/scheduler/interface.go

@@ -40,11 +40,9 @@ type Scheduler interface {
 func NewScheduler(tp string) Scheduler {
 	switch tp {
 	case "table-number":
-		return newTableNumberScheduler(false)
-	case "manual":
-		return newTableNumberScheduler(true)
+		return newTableNumberScheduler()
 	default:
 		log.Info("invalid scheduler type, using default scheduler")
-		return newTableNumberScheduler(true)
+		return newTableNumberScheduler()
 	}
 }

pkg/scheduler/table_number.go

@@ -17,15 +17,13 @@ import "github.com/pingcap/ticdc/cdc/model"
 
 // TableNumberScheduler provides a feature that scheduling by the table number
 type TableNumberScheduler struct {
-	workloads      workloads
-	distributeOnly bool
+	workloads workloads
 }
 
 // newTableNumberScheduler creates a new table number scheduler
-func newTableNumberScheduler(distributeOnly bool) *TableNumberScheduler {
+func newTableNumberScheduler() *TableNumberScheduler {
 	return &TableNumberScheduler{
-		workloads:      make(workloads),
-		distributeOnly: distributeOnly,
+		workloads: make(workloads),
 	}
 }
 
@@ -47,9 +45,6 @@ func (t *TableNumberScheduler) Skewness() float64 {
 // CalRebalanceOperates implements the Scheduler interface
 func (t *TableNumberScheduler) CalRebalanceOperates(targetSkewness float64) (
 	skewness float64, moveTableJobs map[model.TableID]*model.MoveTableJob) {
-	if t.distributeOnly {
-		return 0, nil
-	}
 	var totalTableNumber uint64
 	for _, captureWorkloads := range t.workloads {
 		totalTableNumber += uint64(len(captureWorkloads))

pkg/scheduler/table_number_test.go

@@ -26,7 +26,7 @@ type tableNumberSuite struct{}
 var _ = check.Suite(&tableNumberSuite{})
 
 func (s *tableNumberSuite) TestDistributeTables(c *check.C) {
-	scheduler := newTableNumberScheduler(false)
+	scheduler := newTableNumberScheduler()
 	scheduler.ResetWorkloads("capture1", model.TaskWorkload{
 		1: model.WorkloadInfo{Workload: 1},
 		2: model.WorkloadInfo{Workload: 1}})
@@ -58,7 +58,7 @@ func (s *tableNumberSuite) TestDistributeTables(c *check.C) {
 }
 
 func (s *tableNumberSuite) TestCalRebalanceOperates(c *check.C) {
-	scheduler := newTableNumberScheduler(false)
+	scheduler := newTableNumberScheduler()
 	scheduler.ResetWorkloads("capture1", model.TaskWorkload{
 		1: model.WorkloadInfo{Workload: 1},
 		2: model.WorkloadInfo{Workload: 1}})