cdc: fixes minor bugs pingcap#10168 and pingcap#10169 (pingcap#10170)

close pingcap#10168

Signed-off-by: qupeng <qupeng@pingcap.com>

cdc/kv/client.go

@@ -73,6 +73,8 @@ const (
 	// failed region will be reloaded via `BatchLoadRegionsWithKeyRange` API. So we
 	// don't need to force reload region anymore.
 	regionScheduleReload = false
+
+	scanRegionsConcurrency = 1024
 )
 
 // time interval to force kv client to terminate gRPC stream and reconnect
@@ -432,6 +434,8 @@ func (s *eventFeedSession) eventFeed(ctx context.Context) error {
 	g.Go(func() error { return s.logSlowRegions(ctx) })
 
 	g.Go(func() error {
+		g, ctx := errgroup.WithContext(ctx)
+		g.SetLimit(scanRegionsConcurrency)
 		for {
 			select {
 			case <-ctx.Done():

cdc/processor/sinkmanager/tasks.go

@@ -39,8 +39,8 @@ var (
 	maxUpdateIntervalSize = defaultMaxUpdateIntervalSize
 
 	// Sink manager schedules table tasks based on lag. Limit the max task range
-	// can be helpful to reduce changefeed latency.
-	maxTaskTimeRange = 5 * time.Second
+	// can be helpful to reduce changefeed latency for large initial data.
+	maxTaskTimeRange = 30 * time.Minute
 )
 
 // Used to record the progress of the table.

cdc/processor/sinkmanager/tasks_test.go

@@ -37,7 +37,7 @@ func TestValidateAndAdjustBound(t *testing.T) {
 				StartTs:  439333515018895365,
 				CommitTs: 439333515018895366,
 			},
-			taskTimeRange: 10 * time.Second,
+			taskTimeRange: 60 * time.Minute,
 			expectAdjust:  true,
 		},
 		{